proxygen: folly::ThreadPoolExecutor Class Reference

#include <ThreadPoolExecutor.h>

Inheritance diagram for folly::ThreadPoolExecutor:

Classes
class	Observer

struct	PoolStats

class	StoppedThreadQueue

struct	Task

struct	TaskStats

struct	TaskStatsCallbackRegistry

struct	Thread

class	ThreadHandle

class	ThreadList

Public Types
using	TaskStatsCallback = std::function< void(TaskStats)>

Public Member Functions
	ThreadPoolExecutor (size_t maxThreads, size_t minThreads, std::shared_ptr< ThreadFactory > threadFactory, bool isWaitForAll=false)

	~ThreadPoolExecutor () override

void	add (Func func) override=0

virtual void	add (Func func, std::chrono::milliseconds expiration, Func expireCallback)=0

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

Static Public Member Functions
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)

Protected Types
typedef std::shared_ptr< Thread >	ThreadPtr

Protected Member Functions
void	addThreads (size_t n)

void	removeThreads (size_t n, bool isJoin)

virtual void	threadRun (ThreadPtr thread)=0

virtual void	stopThreads (size_t n)=0

void	joinStoppedThreads (size_t n)

virtual ThreadPtr	makeThread ()

virtual size_t	getPendingTaskCountImpl ()=0

void	ensureActiveThreads ()

void	ensureJoined ()

bool	minActive ()

bool	tryTimeoutThread ()

void	joinKeepAliveOnce ()

Protected Member Functions inherited from folly::DefaultKeepAliveExecutor
void	joinKeepAlive ()

Static Protected Member Functions
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
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}

Additional Inherited Members
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

Detailed Description

Definition at line 55 of file ThreadPoolExecutor.h.

Member Typedef Documentation

using folly::ThreadPoolExecutor::TaskStatsCallback = std::function<void(TaskStats)>

Definition at line 123 of file ThreadPoolExecutor.h.

typedef std::shared_ptr<Thread> folly::ThreadPoolExecutor::ThreadPtr

protected

Definition at line 189 of file ThreadPoolExecutor.h.

Constructor & Destructor Documentation

folly::ThreadPoolExecutor::ThreadPoolExecutor	(	size_t	maxThreads,
		size_t	minThreads,
		std::shared_ptr< ThreadFactory >	threadFactory,
		bool	isWaitForAll = `false`
	)

explicit

Definition at line 37 of file ThreadPoolExecutor.cpp.

References folly::getSyncVecThreadPoolExecutors().

     : threadFactory_(std::move(threadFactory)),
       isWaitForAll_(isWaitForAll),
       taskStatsCallbacks_(std::make_shared<TaskStatsCallbackRegistry>()),
       threadPoolHook_("folly::ThreadPoolExecutor"),
       minThreads_(minThreads),
       threadTimeout_(FLAGS_threadtimeout_ms) {
   getSyncVecThreadPoolExecutors()->push_back(this);
 }

folly::ThreadPoolExecutor::~ThreadPoolExecutor ( )

override

Definition at line 51 of file ThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ThreadList::get(), folly::getSyncVecThreadPoolExecutors(), joinKeepAliveOnce(), and threadList_.

                                         {
   joinKeepAliveOnce();
   CHECK_EQ(0, threadList_.get().size());
   getSyncVecThreadPoolExecutors().withWLock([this](auto& tpe) {
     tpe.erase(std::remove(tpe.begin(), tpe.end(), this), tpe.end());
   });
 }

Member Function Documentation

void folly::ThreadPoolExecutor::add ( Func )

overridepure virtual

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

Implements folly::Executor.

Implemented in folly::CPUThreadPoolExecutor, and folly::IOThreadPoolExecutor.

virtual void folly::ThreadPoolExecutor::add	(	Func	func,
		std::chrono::milliseconds	expiration,
		Func	expireCallback
	)

pure virtual

Implemented in folly::CPUThreadPoolExecutor, and folly::IOThreadPoolExecutor.

void folly::ThreadPoolExecutor::addObserver ( std::shared_ptr< Observer > o )

Definition at line 349 of file ThreadPoolExecutor.cpp.

References activeThreads_, ensureActiveThreads(), folly::ThreadPoolExecutor::ThreadList::get(), maxThreads_, observers_, threadList_, and threadListLock_.

Referenced by TEST(), and folly::ThreadPoolExecutor::Observer::threadNotYetStopped().

                                                               {
   {
     SharedMutex::WriteHolder r{&threadListLock_};
     observers_.push_back(o);
     for (auto& thread : threadList_.get()) {
       o->threadPreviouslyStarted(thread.get());
     }
   }
   while (activeThreads_.load(std::memory_order_relaxed) <
          maxThreads_.load(std::memory_order_relaxed)) {
     ensureActiveThreads();
   }
 }

void folly::ThreadPoolExecutor::addThreads ( size_t n )

protected

Definition at line 180 of file ThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ThreadList::add(), folly::netops::bind(), i, makeThread(), observers_, threadFactory_, threadList_, and threadRun().

Referenced by ensureActiveThreads(), setNumThreads(), and setThreadDeathTimeout().

                                             {
   std::vector<ThreadPtr> newThreads;
   for (size_t i = 0; i < n; i++) {
     newThreads.push_back(makeThread());
   }
   for (auto& thread : newThreads) {
     // TODO need a notion of failing to create the thread
     // and then handling for that case
     thread->handle = threadFactory_->newThread(
         std::bind(&ThreadPoolExecutor::threadRun, this, thread));
     threadList_.add(thread);
   }
   for (auto& thread : newThreads) {
     thread->startupBaton.wait();
   }
   for (auto& o : observers_) {
     for (auto& thread : newThreads) {
       o->threadStarted(thread.get());
     }
   }
 }

void folly::ThreadPoolExecutor::ensureActiveThreads ( )

protected

Definition at line 434 of file ThreadPoolExecutor.cpp.

References activeThreads_, addThreads(), folly::asymmetricLightBarrier(), ensureJoined(), maxThreads_, and threadListLock_.

Referenced by folly::IOThreadPoolExecutor::add(), folly::CPUThreadPoolExecutor::add(), addObserver(), and folly::IOThreadPoolExecutor::getEventBase().

                                              {
   ensureJoined();
 
   // Matches barrier in tryTimeoutThread().  Ensure task added
   // is seen before loading activeThreads_ below.
   asymmetricLightBarrier();
 
   // Fast path assuming we are already at max threads.
   auto active = activeThreads_.load(std::memory_order_relaxed);
   auto total = maxThreads_.load(std::memory_order_relaxed);
 
   if (active >= total) {
     return;
   }
 
   SharedMutex::WriteHolder w{&threadListLock_};
   // Double check behind lock.
   active = activeThreads_.load(std::memory_order_relaxed);
   total = maxThreads_.load(std::memory_order_relaxed);
   if (active >= total) {
     return;
   }
   ThreadPoolExecutor::addThreads(1);
   activeThreads_.store(active + 1, std::memory_order_relaxed);
 }

void folly::ThreadPoolExecutor::ensureJoined ( )

protected

Definition at line 380 of file ThreadPoolExecutor.cpp.

References joinStoppedThreads(), threadListLock_, and threadsToJoin_.

Referenced by ensureActiveThreads().

                                       {
   auto tojoin = threadsToJoin_.load(std::memory_order_relaxed);
   if (tojoin) {
     {
       SharedMutex::WriteHolder w{&threadListLock_};
       tojoin = threadsToJoin_.load(std::memory_order_relaxed);
       threadsToJoin_.store(0, std::memory_order_relaxed);
     }
     joinStoppedThreads(tojoin);
   }
 }

std::string folly::ThreadPoolExecutor::getName ( )

Definition at line 287 of file ThreadPoolExecutor.cpp.

References folly::demangle(), folly::ThreadPoolExecutor::Thread::nextId, threadFactory_, and folly::toStdString().

                                       {
   auto ntf = dynamic_cast<NamedThreadFactory*>(threadFactory_.get());
   if (ntf == nullptr) {
     return folly::demangle(typeid(*this).name()).toStdString();
   }
 
   return ntf->getNamePrefix();
 }

size_t folly::ThreadPoolExecutor::getPendingTaskCount ( )

Definition at line 282 of file ThreadPoolExecutor.cpp.

References getPendingTaskCountImpl(), and threadListLock_.

                                                {
   SharedMutex::ReadHolder r{&threadListLock_};
   return getPendingTaskCountImpl();
 }

virtual size_t folly::ThreadPoolExecutor::getPendingTaskCountImpl ( )

protectedpure virtual

Implemented in folly::CPUThreadPoolExecutor, and folly::IOThreadPoolExecutor.

Referenced by getPendingTaskCount(), getPoolStats(), makeThread(), setNumThreads(), and tryTimeoutThread().

ThreadPoolExecutor::PoolStats folly::ThreadPoolExecutor::getPoolStats ( )

Definition at line 257 of file ThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::PoolStats::activeThreadCount, activeThreads_, folly::ThreadPoolExecutor::ThreadList::get(), getPendingTaskCountImpl(), folly::ThreadPoolExecutor::PoolStats::idleThreadCount, max, folly::ThreadPoolExecutor::PoolStats::maxIdleTime, maxThreads_, now(), folly::ThreadPoolExecutor::PoolStats::pendingTaskCount, folly::ThreadPoolExecutor::PoolStats::threadCount, threadList_, threadListLock_, and folly::ThreadPoolExecutor::PoolStats::totalTaskCount.

Referenced by TEST().

                                                              {
   const auto now = std::chrono::steady_clock::now();
   SharedMutex::ReadHolder r{&threadListLock_};
   ThreadPoolExecutor::PoolStats stats;
   size_t activeTasks = 0;
   size_t idleAlive = 0;
   for (auto thread : threadList_.get()) {
     if (thread->idle) {
       const std::chrono::nanoseconds idleTime = now - thread->lastActiveTime;
       stats.maxIdleTime = std::max(stats.maxIdleTime, idleTime);
       idleAlive++;
     } else {
       activeTasks++;
     }
   }
   stats.pendingTaskCount = getPendingTaskCountImpl();
   stats.totalTaskCount = stats.pendingTaskCount + activeTasks;
 
   stats.threadCount = maxThreads_.load(std::memory_order_relaxed);
   stats.activeThreadCount =
       activeThreads_.load(std::memory_order_relaxed) - idleAlive;
   stats.idleThreadCount = stats.threadCount - stats.activeThreadCount;
   return stats;
 }

std::shared_ptr<ThreadFactory> folly::ThreadPoolExecutor::getThreadFactory ( )

inline

Definition at line 74 of file ThreadPoolExecutor.h.

References f, join(), numActiveThreads(), numThreads(), setNumThreads(), stop(), threadFactory_, and withAll().

                                                   {
     return threadFactory_;
   }

void folly::ThreadPoolExecutor::join ( )

Definition at line 232 of file ThreadPoolExecutor.cpp.

References activeThreads_, folly::ThreadPoolExecutor::ThreadList::get(), joinKeepAliveOnce(), joinStoppedThreads(), maxThreads_, removeThreads(), folly::ThreadPoolExecutor::StoppedThreadQueue::size(), stoppedThreads_, threadList_, threadListLock_, and threadsToJoin_.

Referenced by getThreadFactory(), and TEST().

                               {
   joinKeepAliveOnce();
   size_t n = 0;
   {
     SharedMutex::WriteHolder w{&threadListLock_};
     maxThreads_.store(0, std::memory_order_release);
     activeThreads_.store(0, std::memory_order_release);
     n = threadList_.get().size();
     removeThreads(n, true);
     n += threadsToJoin_.load(std::memory_order_relaxed);
     threadsToJoin_.store(0, std::memory_order_relaxed);
   }
   joinStoppedThreads(n);
   CHECK_EQ(0, threadList_.get().size());
   CHECK_EQ(0, stoppedThreads_.size());
 }

void folly::ThreadPoolExecutor::joinKeepAliveOnce ( )

inlineprotected

Definition at line 318 of file ThreadPoolExecutor.h.

References folly::exchange(), folly::DefaultKeepAliveExecutor::joinKeepAlive(), and keepAliveJoined_.

Referenced by join(), stop(), and ~ThreadPoolExecutor().

                            {
     if (!std::exchange(keepAliveJoined_, true)) {
       joinKeepAlive();
     }
   }

void folly::ThreadPoolExecutor::joinStoppedThreads ( size_t n )

protected

Definition at line 208 of file ThreadPoolExecutor.cpp.

References i, stoppedThreads_, and folly::ThreadPoolExecutor::StoppedThreadQueue::take().

Referenced by ensureJoined(), join(), setNumThreads(), and stop().

                                                     {
   for (size_t i = 0; i < n; i++) {
     auto thread = stoppedThreads_.take();
     thread->handle.join();
   }
 }

virtual ThreadPtr folly::ThreadPoolExecutor::makeThread ( )

inlineprotectedvirtual

Reimplemented in folly::IOThreadPoolExecutor.

Definition at line 222 of file ThreadPoolExecutor.h.

References getPendingTaskCountImpl().

Referenced by addThreads().

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

bool folly::ThreadPoolExecutor::minActive ( )

protected

Definition at line 462 of file ThreadPoolExecutor.cpp.

References activeThreads_, and minThreads_.

Referenced by tryTimeoutThread().

                                    {
   return activeThreads_.load(std::memory_order_relaxed) >
       minThreads_.load(std::memory_order_relaxed);
 }

size_t folly::ThreadPoolExecutor::numActiveThreads ( )

Definition at line 120 of file ThreadPoolExecutor.cpp.

References activeThreads_.

Referenced by getThreadFactory().

                                             {
   return activeThreads_.load(std::memory_order_relaxed);
 }

size_t folly::ThreadPoolExecutor::numThreads ( )

Definition at line 116 of file ThreadPoolExecutor.cpp.

References maxThreads_.

Referenced by getThreadFactory(), setNumThreads(), and setThreadFactory().

                                       {
   return maxThreads_.load(std::memory_order_relaxed);
 }

void folly::ThreadPoolExecutor::removeObserver ( std::shared_ptr< Observer > o )

Definition at line 363 of file ThreadPoolExecutor.cpp.

References folly::ThreadPoolExecutor::ThreadList::get(), observers_, threadList_, and threadListLock_.

Referenced by TEST(), and folly::ThreadPoolExecutor::Observer::threadNotYetStopped().

                                                                  {
   SharedMutex::WriteHolder r{&threadListLock_};
   for (auto& thread : threadList_.get()) {
     o->threadNotYetStopped(thread.get());
   }
 
   for (auto it = observers_.begin(); it != observers_.end(); it++) {
     if (*it == o) {
       observers_.erase(it);
       return;
     }
   }
   DCHECK(false);
 }

void folly::ThreadPoolExecutor::removeThreads	(	size_t	n,
		bool	isJoin
	)

protected

Definition at line 203 of file ThreadPoolExecutor.cpp.

References isJoin_, and stopThreads().

Referenced by join(), setNumThreads(), setThreadDeathTimeout(), and stop().

                                                             {
   isJoin_ = isJoin;
   stopThreads(n);
 }

void folly::ThreadPoolExecutor::runTask	(	const ThreadPtr &	thread,
		Task &&	task
	)

staticprotected

Definition at line 71 of file ThreadPoolExecutor.cpp.

References name, now(), and SCOPE_EXIT.

Referenced by folly::IOThreadPoolExecutor::add(), and folly::CPUThreadPoolExecutor::threadRun().

                                                                      {
   thread->idle = false;
   auto startTime = std::chrono::steady_clock::now();
   task.stats_.waitTime = startTime - task.enqueueTime_;
   if (task.expiration_ > std::chrono::milliseconds(0) &&
       task.stats_.waitTime >= task.expiration_) {
     task.stats_.expired = true;
     if (task.expireCallback_ != nullptr) {
       task.expireCallback_();
     }
   } else {
     folly::RequestContextScopeGuard rctx(task.context_);
     try {
       task.func_();
     } catch (const std::exception& e) {
       LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled "
                  << typeid(e).name() << " exception: " << e.what();
     } catch (...) {
       LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled non-exception "
                     "object";
     }
     task.stats_.runTime = std::chrono::steady_clock::now() - startTime;
   }
   thread->idle = true;
   thread->lastActiveTime = std::chrono::steady_clock::now();
   thread->taskStatsCallbacks->callbackList.withRLock([&](auto& callbacks) {
     *thread->taskStatsCallbacks->inCallback = true;
     SCOPE_EXIT {
       *thread->taskStatsCallbacks->inCallback = false;
     };
     try {
       for (auto& callback : callbacks) {
         callback(task.stats_);
       }
     } catch (const std::exception& e) {
       LOG(ERROR) << "ThreadPoolExecutor: task stats callback threw "
                     "unhandled "
                  << typeid(e).name() << " exception: " << e.what();
     } catch (...) {
       LOG(ERROR) << "ThreadPoolExecutor: task stats callback threw "
                     "unhandled non-exception object";
     }
   });
 }

void folly::ThreadPoolExecutor::setNumThreads ( size_t numThreads )

Definition at line 125 of file ThreadPoolExecutor.cpp.

References activeThreads_, addThreads(), getPendingTaskCountImpl(), joinStoppedThreads(), maxThreads_, min, minThreads_, numThreads(), observers_, removeThreads(), and threadListLock_.

Referenced by folly::CPUThreadPoolExecutor::CPUThreadPoolExecutor(), getThreadFactory(), folly::IOThreadPoolExecutor::IOThreadPoolExecutor(), and TEST().

                                                         {
   /* Since ThreadPoolExecutor may be dynamically adjusting the number of
      threads, we adjust the relevant variables instead of changing
      the number of threads directly.  Roughly:
 
      If numThreads < minthreads reset minThreads to numThreads.
 
      If numThreads < active threads, reduce number of running threads.
 
      If the number of pending tasks is > 0, then increase the currently
      active number of threads such that we can run all the tasks, or reach
      numThreads.
 
      Note that if there are observers, we actually have to create all
      the threads, because some observer implementations need to 'observe'
      all thread creation (see tests for an example of this)
   */
 
   size_t numThreadsToJoin = 0;
   {
     SharedMutex::WriteHolder w{&threadListLock_};
     auto pending = getPendingTaskCountImpl();
     maxThreads_.store(numThreads, std::memory_order_relaxed);
     auto active = activeThreads_.load(std::memory_order_relaxed);
     auto minthreads = minThreads_.load(std::memory_order_relaxed);
     if (numThreads < minthreads) {
       minthreads = numThreads;
       minThreads_.store(numThreads, std::memory_order_relaxed);
     }
     if (active > numThreads) {
       numThreadsToJoin = active - numThreads;
       if (numThreadsToJoin > active - minthreads) {
         numThreadsToJoin = active - minthreads;
       }
       ThreadPoolExecutor::removeThreads(numThreadsToJoin, false);
       activeThreads_.store(
           active - numThreadsToJoin, std::memory_order_relaxed);
     } else if (pending > 0 || observers_.size() > 0 || active < minthreads) {
       size_t numToAdd = std::min(pending, numThreads - active);
       if (observers_.size() > 0) {
         numToAdd = numThreads - active;
       }
       if (active + numToAdd < minthreads) {
         numToAdd = minthreads - active;
       }
       ThreadPoolExecutor::addThreads(numToAdd);
       activeThreads_.store(active + numToAdd, std::memory_order_relaxed);
     }
   }
 
   /* We may have removed some threads, attempt to join them */
   joinStoppedThreads(numThreadsToJoin);
 }

void folly::ThreadPoolExecutor::setThreadDeathTimeout ( std::chrono::milliseconds timeout )

inline

Definition at line 157 of file ThreadPoolExecutor.h.

References addThreads(), removeThreads(), and threadTimeout_.

Referenced by TEST().

                                                             {
     threadTimeout_ = timeout;
   }

void folly::ThreadPoolExecutor::setThreadFactory ( std::shared_ptr< ThreadFactory > threadFactory )

inline

Definition at line 69 of file ThreadPoolExecutor.h.

References folly::gen::move, numThreads(), and threadFactory_.

                                                                     {
     CHECK(numThreads() == 0);
     threadFactory_ = std::move(threadFactory);
   }

void folly::ThreadPoolExecutor::stop ( )

Definition at line 215 of file ThreadPoolExecutor.cpp.

Referenced by getThreadFactory(), stop< IOThreadPoolExecutor >(), folly::CPUThreadPoolExecutor::~CPUThreadPoolExecutor(), and folly::IOThreadPoolExecutor::~IOThreadPoolExecutor().

                               {
   joinKeepAliveOnce();
   size_t n = 0;
   {
     SharedMutex::WriteHolder w{&threadListLock_};
     maxThreads_.store(0, std::memory_order_release);
     activeThreads_.store(0, std::memory_order_release);
     n = threadList_.get().size();
     removeThreads(n, false);
     n += threadsToJoin_.load(std::memory_order_relaxed);
     threadsToJoin_.store(0, std::memory_order_relaxed);
   }
   joinStoppedThreads(n);
   CHECK_EQ(0, threadList_.get().size());
   CHECK_EQ(0, stoppedThreads_.size());
 }

virtual void folly::ThreadPoolExecutor::stopThreads ( size_t n )

protectedpure virtual

Implemented in folly::CPUThreadPoolExecutor, and folly::IOThreadPoolExecutor.

Referenced by removeThreads().

void folly::ThreadPoolExecutor::subscribeToTaskStats ( TaskStatsCallback cb )

Definition at line 298 of file ThreadPoolExecutor.cpp.

References folly::gen::move, and taskStatsCallbacks_.

                                                                   {
   if (*taskStatsCallbacks_->inCallback) {
     throw std::runtime_error("cannot subscribe in task stats callback");
   }
   taskStatsCallbacks_->callbackList.wlock()->push_back(std::move(cb));
 }

virtual void folly::ThreadPoolExecutor::threadRun ( ThreadPtr thread )

protectedpure virtual

Implemented in folly::CPUThreadPoolExecutor, and folly::IOThreadPoolExecutor.

Referenced by addThreads().

bool folly::ThreadPoolExecutor::tryTimeoutThread ( )

protected

Definition at line 393 of file ThreadPoolExecutor.cpp.

References activeThreads_, folly::asymmetricHeavyBarrier(), getPendingTaskCountImpl(), minActive(), and threadsToJoin_.

Referenced by folly::CPUThreadPoolExecutor::taskShouldStop().

                                           {
   // Try to stop based on idle thread timeout (try_take_for),
   // if there are at least minThreads running.
   if (!minActive()) {
     return false;
   }
 
   // Remove thread from active count
   activeThreads_.store(
       activeThreads_.load(std::memory_order_relaxed) - 1,
       std::memory_order_relaxed);
 
   // There is a memory ordering constraint w.r.t the queue
   // implementation's add() and getPendingTaskCountImpl() - while many
   // queues have seq_cst ordering, some do not, so add an explicit
   // barrier.  tryTimeoutThread is the slow path and only happens once
   // every thread timeout; use asymmetric barrier to keep add() fast.
   asymmetricHeavyBarrier();
 
   // If this is based on idle thread timeout, then
   // adjust vars appropriately (otherwise stop() or join()
   // does this).
   if (getPendingTaskCountImpl() > 0) {
     // There are still pending tasks, we can't stop yet.
     // re-up active threads and return.
     activeThreads_.store(
         activeThreads_.load(std::memory_order_relaxed) + 1,
         std::memory_order_relaxed);
     return false;
   }
 
   threadsToJoin_.store(
       threadsToJoin_.load(std::memory_order_relaxed) + 1,
       std::memory_order_relaxed);
 
   return true;
 }

void folly::ThreadPoolExecutor::withAll ( FunctionRef< void(ThreadPoolExecutor &)> f )

static

Execute f against all ThreadPoolExecutors, primarily for retrieving and exporting stats.

Definition at line 249 of file ThreadPoolExecutor.cpp.

References f, and folly::getSyncVecThreadPoolExecutors().

Referenced by getNumThreadPoolExecutors(), and getThreadFactory().

                                                                          {
   getSyncVecThreadPoolExecutors().withRLock([f](auto& tpes) {
     for (auto tpe : tpes) {
       f(*tpe);
     }
   });
 }