folly::IOThreadPoolExecutor Class Reference

#include <IOThreadPoolExecutor.h>

Inheritance diagram for folly::IOThreadPoolExecutor:

Classes
struct	IOThread

Public Member Functions
	IOThreadPoolExecutor (size_t numThreads, std::shared_ptr< ThreadFactory > threadFactory=std::make_shared< NamedThreadFactory >("IOThreadPool"), folly::EventBaseManager *ebm=folly::EventBaseManager::get(), bool waitForAll=false)
	~IOThreadPoolExecutor () override
void	add (Func func) override
void	add (Func func, std::chrono::milliseconds expiration, Func expireCallback=nullptr) override
folly::EventBase *	getEventBase () override
folly::EventBaseManager *	getEventBaseManager ()
Public Member Functions inherited from folly::ThreadPoolExecutor
	ThreadPoolExecutor (size_t maxThreads, size_t minThreads, std::shared_ptr< ThreadFactory > threadFactory, bool isWaitForAll=false)
	~ThreadPoolExecutor () override
void	setThreadFactory (std::shared_ptr< ThreadFactory > threadFactory)
std::shared_ptr< ThreadFactory >	getThreadFactory ()
size_t	numThreads ()
void	setNumThreads (size_t numThreads)
size_t	numActiveThreads ()
void	stop ()
void	join ()
PoolStats	getPoolStats ()
size_t	getPendingTaskCount ()
std::string	getName ()
void	subscribeToTaskStats (TaskStatsCallback cb)
void	addObserver (std::shared_ptr< Observer >)
void	removeObserver (std::shared_ptr< Observer >)
void	setThreadDeathTimeout (std::chrono::milliseconds timeout)
Public Member Functions inherited from folly::DefaultKeepAliveExecutor
	DefaultKeepAliveExecutor ()
virtual	~DefaultKeepAliveExecutor ()
folly::Executor::KeepAlive	weakRef ()
Public Member Functions inherited from folly::Executor
virtual	~Executor ()
virtual void	addWithPriority (Func, int8_t priority)
virtual uint8_t	getNumPriorities () const
Public Member Functions inherited from folly::IOExecutor
	~IOExecutor () override=default

Static Public Member Functions
static folly::EventBase *	getEventBase (ThreadPoolExecutor::ThreadHandle *)
Static Public Member Functions inherited from folly::ThreadPoolExecutor
static void	withAll (FunctionRef< void(ThreadPoolExecutor &)> f)
Static Public Member Functions inherited from folly::Executor
template<typename ExecutorT >
static KeepAlive< ExecutorT >	getKeepAliveToken (ExecutorT *executor)
template<typename ExecutorT >
static KeepAlive< ExecutorT >	getKeepAliveToken (ExecutorT &executor)

Private Member Functions
ThreadPtr	makeThread () override
std::shared_ptr< IOThread >	pickThread ()
void	threadRun (ThreadPtr thread) override
void	stopThreads (size_t n) override
size_t	getPendingTaskCountImpl () override

Private Attributes
std::atomic< size_t >	nextThread_
folly::ThreadLocal< std::shared_ptr< IOThread > >	thisThread_
folly::EventBaseManager *	eventBaseManager_

Additional Inherited Members
Public Types inherited from folly::ThreadPoolExecutor
using	TaskStatsCallback = std::function< void(TaskStats)>
Static Public Attributes inherited from folly::Executor
static const int8_t	LO_PRI = SCHAR_MIN
static const int8_t	MID_PRI = 0
static const int8_t	HI_PRI = SCHAR_MAX
Protected Types inherited from folly::ThreadPoolExecutor
typedef std::shared_ptr< Thread >	ThreadPtr
Protected Member Functions inherited from folly::ThreadPoolExecutor
void	addThreads (size_t n)
void	removeThreads (size_t n, bool isJoin)
void	joinStoppedThreads (size_t n)
void	ensureActiveThreads ()
void	ensureJoined ()
bool	minActive ()
bool	tryTimeoutThread ()
void	joinKeepAliveOnce ()
Protected Member Functions inherited from folly::DefaultKeepAliveExecutor
void	joinKeepAlive ()
Static Protected Member Functions inherited from folly::ThreadPoolExecutor
static void	runTask (const ThreadPtr &thread, Task &&task)
Static Protected Member Functions inherited from folly::Executor
template<typename ExecutorT >
static bool	isKeepAliveDummy (const KeepAlive< ExecutorT > &keepAlive)
template<typename ExecutorT >
static KeepAlive< ExecutorT >	makeKeepAlive (ExecutorT *executor)
Protected Attributes inherited from folly::ThreadPoolExecutor
std::shared_ptr< ThreadFactory >	threadFactory_
const bool	isWaitForAll_
ThreadList	threadList_
SharedMutex	threadListLock_
StoppedThreadQueue	stoppedThreads_
std::atomic< bool >	isJoin_ {false}
std::shared_ptr< TaskStatsCallbackRegistry >	taskStatsCallbacks_
std::vector< std::shared_ptr< Observer > >	observers_
folly::ThreadPoolListHook	threadPoolHook_
std::atomic< size_t >	maxThreads_ {0}
std::atomic< size_t >	minThreads_ {0}
std::atomic< size_t >	activeThreads_ {0}
std::atomic< size_t >	threadsToJoin_ {0}
std::chrono::milliseconds	threadTimeout_ {0}
bool	keepAliveJoined_ {false}

Detailed Description

A Thread Pool for IO bound tasks

Note

Uses event_fd for notification, and waking an epoll loop. There is one queue (NotificationQueue specifically) per thread/epoll. If the thread is already running and not waiting on epoll, we don't make any additional syscalls to wake up the loop, just put the new task in the queue. If any thread has been waiting for more than a few seconds, its stack is madvised away. Currently however tasks are scheduled round robin on the queues, so unless there is no work going on, this isn't very effective. Since there is one queue per thread, there is hardly any contention on the queues - so a simple spinlock around an std::deque is used for the tasks. There is no max queue size. By default, there is one thread per core - it usually doesn't make sense to have more IO threads than this, assuming they don't block.

getEventBase() will return an EventBase you can schedule IO work on directly, chosen round-robin.

N.B. For this thread pool, stop() behaves like join() because outstanding tasks belong to the event base and will be executed upon its destruction.

Definition at line 52 of file IOThreadPoolExecutor.h.

Constructor & Destructor Documentation

folly::IOThreadPoolExecutor::IOThreadPoolExecutor ( size_t  numThreads, std::shared_ptr< ThreadFactory >  threadFactory = std::make_shared<NamedThreadFactory>("IOThreadPool"), folly::EventBaseManager *  ebm = folly::EventBaseManager::get(), bool  waitForAll = false  )

explicit

Definition at line 70 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::setNumThreads().

    : ThreadPoolExecutor(
          numThreads,
          FLAGS_dynamic_iothreadpoolexecutor ? 0 : numThreads,
          std::move(threadFactory),
          waitForAll),
      nextThread_(0),
      eventBaseManager_(ebm) {
  setNumThreads(numThreads);
}

folly::IOThreadPoolExecutor::~IOThreadPoolExecutor ( )

override

Definition at line 85 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::stop().

                                            {
  stop();
}

Member Function Documentation

void folly::IOThreadPoolExecutor::add ( Func  )

overridevirtual

Enqueue a function to executed by this executor. This and all variants must be threadsafe.

Implements folly::ThreadPoolExecutor.

Definition at line 89 of file IOThreadPoolExecutor.cpp.

References folly::gen::move.

Referenced by destroy< IOThreadPoolExecutor >(), and stop< IOThreadPoolExecutor >().

                                        {
  add(std::move(func), std::chrono::milliseconds(0));
}

void folly::IOThreadPoolExecutor::add ( Func  func, std::chrono::milliseconds  expiration, Func  expireCallback = nullptr  )

overridevirtual

Implements folly::ThreadPoolExecutor.

Definition at line 93 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ensureActiveThreads(), folly::ThreadPoolExecutor::ThreadList::get(), folly::gen::move, pickThread(), folly::ThreadPoolExecutor::runTask(), folly::ThreadPoolExecutor::threadList_, and folly::ThreadPoolExecutor::threadListLock_.

                         {
  ensureActiveThreads();
  SharedMutex::ReadHolder r{&threadListLock_};
  if (threadList_.get().empty()) {
    throw std::runtime_error("No threads available");
  }
  auto ioThread = pickThread();

  auto task = Task(std::move(func), expiration, std::move(expireCallback));
  auto wrappedFunc = [ioThread, task = std::move(task)]() mutable {
    runTask(ioThread, std::move(task));
    ioThread->pendingTasks--;
  };

  ioThread->pendingTasks++;
  if (!ioThread->eventBase->runInEventBaseThread(std::move(wrappedFunc))) {
    ioThread->pendingTasks--;
    throw std::runtime_error("Unable to run func in event base thread");
  }
}

EventBase * folly::IOThreadPoolExecutor::getEventBase ( )

overridevirtual

Implements folly::IOExecutor.

Definition at line 144 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ensureActiveThreads(), pickThread(), and folly::ThreadPoolExecutor::threadListLock_.

Referenced by wangle::ServerWorkerPool::threadStarted().

                                              {
  ensureActiveThreads();
  SharedMutex::ReadHolder r{&threadListLock_};
  return pickThread()->eventBase;
}

EventBase * folly::IOThreadPoolExecutor::getEventBase ( ThreadPoolExecutor::ThreadHandle *  h )

static

Definition at line 150 of file IOThreadPoolExecutor.cpp.

References folly::IOThreadPoolExecutor::IOThread::eventBase, and h.

                                       {
  auto thread = dynamic_cast<IOThread*>(h);

  if (thread) {
    return thread->eventBase;
  }

  return nullptr;
}

EventBaseManager * folly::IOThreadPoolExecutor::getEventBaseManager

(

)

Definition at line 161 of file IOThreadPoolExecutor.cpp.

References eventBaseManager_.

                                                            {
  return eventBaseManager_;
}

size_t folly::IOThreadPoolExecutor::getPendingTaskCountImpl ( )

overrideprivatevirtual

Implements folly::ThreadPoolExecutor.

Definition at line 225 of file IOThreadPoolExecutor.cpp.

References count, folly::ThreadPoolExecutor::ThreadList::get(), folly::IOThreadPoolExecutor::IOThread::pendingTasks, and folly::ThreadPoolExecutor::threadList_.

                                                     {
  size_t count = 0;
  for (const auto& thread : threadList_.get()) {
    auto ioThread = std::static_pointer_cast<IOThread>(thread);
    size_t pendingTasks = ioThread->pendingTasks;
    if (pendingTasks > 0 && !ioThread->idle) {
      pendingTasks--;
    }
    count += pendingTasks;
  }
  return count;
}

std::shared_ptr< ThreadPoolExecutor::Thread > folly::IOThreadPoolExecutor::makeThread ( )

overrideprivatevirtual

Reimplemented from folly::ThreadPoolExecutor.

Definition at line 165 of file IOThreadPoolExecutor.cpp.

                                                                         {
  return std::make_shared<IOThread>(this);
}

std::shared_ptr< IOThreadPoolExecutor::IOThread > folly::IOThreadPoolExecutor::pickThread ( )

private

Definition at line 118 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ThreadList::get(), nextThread_, thisThread_, and folly::ThreadPoolExecutor::threadList_.

Referenced by add(), and getEventBase().

                                 {
  auto& me = *thisThread_;
  auto& ths = threadList_.get();
  // When new task is added to IOThreadPoolExecutor, a thread is chosen for it
  // to be executed on, thisThread_ is by default chosen, however, if the new
  // task is added by the clean up operations on thread destruction, thisThread_
  // is not an available thread anymore, thus, always check whether or not
  // thisThread_ is an available thread before choosing it.
  if (me && std::find(ths.cbegin(), ths.cend(), me) != ths.cend()) {
    return me;
  }
  auto n = ths.size();
  if (n == 0) {
    // XXX I think the only way this can happen is if somebody calls
    // getEventBase (1) from one of the executor's threads while the executor
    // is stopping or getting downsized to zero or (2) from outside the executor
    // when it has no threads. In the first case, it's not obvious what the
    // correct behavior should be-- do we really want to return ourselves even
    // though we're about to exit? (The comment above seems to imply no.) In
    // the second case, `!me` so we'll crash anyway.
    return me;
  }
  auto thread = ths[nextThread_.fetch_add(1, std::memory_order_relaxed) % n];
  return std::static_pointer_cast<IOThread>(thread);
}

void folly::IOThreadPoolExecutor::stopThreads ( size_t  n )

overrideprivatevirtual

Implements folly::ThreadPoolExecutor.

Definition at line 202 of file IOThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::StoppedThreadQueue::add(), folly::ThreadPoolExecutor::ThreadList::get(), folly::gen::guard(), i, folly::ThreadPoolExecutor::observers_, folly::ThreadPoolExecutor::ThreadList::remove(), folly::IOThreadPoolExecutor::IOThread::shouldRun, folly::ThreadPoolExecutor::stoppedThreads_, and folly::ThreadPoolExecutor::threadList_.

                                               {
  std::vector<ThreadPtr> stoppedThreads;
  stoppedThreads.reserve(n);
  for (size_t i = 0; i < n; i++) {
    const auto ioThread =
        std::static_pointer_cast<IOThread>(threadList_.get()[i]);
    for (auto& o : observers_) {
      o->threadStopped(ioThread.get());
    }
    ioThread->shouldRun = false;
    stoppedThreads.push_back(ioThread);
    std::lock_guard<std::mutex> guard(ioThread->eventBaseShutdownMutex_);
    if (ioThread->eventBase) {
      ioThread->eventBase->terminateLoopSoon();
    }
  }
  for (auto thread : stoppedThreads) {
    stoppedThreads_.add(thread);
    threadList_.remove(thread);
  }
}

void folly::IOThreadPoolExecutor::threadRun ( ThreadPtr  thread )

overrideprivatevirtual

Implements folly::ThreadPoolExecutor.

Definition at line 169 of file IOThreadPoolExecutor.cpp.

References folly::EventBaseManager::clearEventBase(), folly::IOThreadPoolExecutor::IOThread::eventBase, eventBaseManager_, folly::EventBaseManager::getEventBase(), folly::gen::guard(), folly::ThreadPoolExecutor::isJoin_, folly::ThreadPoolExecutor::isWaitForAll_, folly::ThreadPoolListHook::registerThread(), thisThread_, and folly::ThreadPoolExecutor::threadPoolHook_.

                                                     {
  this->threadPoolHook_.registerThread();

  const auto ioThread = std::static_pointer_cast<IOThread>(thread);
  ioThread->eventBase = eventBaseManager_->getEventBase();
  thisThread_.reset(new std::shared_ptr<IOThread>(ioThread));

  auto idler = std::make_unique<MemoryIdlerTimeout>(ioThread->eventBase);
  ioThread->eventBase->runBeforeLoop(idler.get());

  ioThread->eventBase->runInEventBaseThread(
      [thread] { thread->startupBaton.post(); });
  while (ioThread->shouldRun) {
    ioThread->eventBase->loopForever();
  }
  if (isJoin_) {
    while (ioThread->pendingTasks > 0) {
      ioThread->eventBase->loopOnce();
    }
  }
  idler.reset();
  if (isWaitForAll_) {
    // some tasks, like thrift asynchronous calls, create additional
    // event base hookups, let's wait till all of them complete.
    ioThread->eventBase->loop();
  }

  std::lock_guard<std::mutex> guard(ioThread->eventBaseShutdownMutex_);
  ioThread->eventBase = nullptr;
  eventBaseManager_->clearEventBase();
}

Member Data Documentation

folly::EventBaseManager* folly::IOThreadPoolExecutor::eventBaseManager_

private

Definition at line 94 of file IOThreadPoolExecutor.h.

Referenced by getEventBaseManager(), and threadRun().

std::atomic<size_t> folly::IOThreadPoolExecutor::nextThread_

private

Definition at line 92 of file IOThreadPoolExecutor.h.

Referenced by pickThread().

folly::ThreadLocal<std::shared_ptr<IOThread> > folly::IOThreadPoolExecutor::thisThread_

private

Definition at line 93 of file IOThreadPoolExecutor.h.

Referenced by pickThread(), and threadRun().

---

The documentation for this class was generated from the following files:

• proxygen/folly/folly/executors/IOThreadPoolExecutor.h
• proxygen/folly/folly/executors/IOThreadPoolExecutor.cpp