proxygen: folly::ThreadedExecutor Class Reference

#include <ThreadedExecutor.h>

Inheritance diagram for folly::ThreadedExecutor:

Public Member Functions
	ThreadedExecutor (std::shared_ptr< ThreadFactory > threadFactory=newDefaultThreadFactory())

	~ThreadedExecutor () override

	ThreadedExecutor (ThreadedExecutor const &)=delete

	ThreadedExecutor (ThreadedExecutor &&)=delete

ThreadedExecutor &	operator= (ThreadedExecutor const &)=delete

ThreadedExecutor &	operator= (ThreadedExecutor &&)=delete

void	add (Func func) override

Public Member Functions inherited from folly::Executor
virtual	~Executor ()

virtual void	addWithPriority (Func, int8_t priority)

virtual uint8_t	getNumPriorities () const

Private Member Functions
void	notify ()

void	control ()

void	controlWait ()

bool	controlPerformAll ()

void	controlJoinFinishedThreads ()

void	controlLaunchEnqueuedTasks ()

void	work (Func &func)

Static Private Member Functions
static std::shared_ptr< ThreadFactory >	newDefaultThreadFactory ()

Private Attributes
std::shared_ptr< ThreadFactory >	threadFactory_

std::atomic< bool >	stopping_ {false}

std::mutex	controlm_

std::condition_variable	controlc_

bool	controls_ = false

std::thread	controlt_

std::mutex	enqueuedm_

std::deque< Func >	enqueued_

std::map< std::thread::id, std::thread >	running_

std::mutex	finishedm_

std::deque< std::thread::id >	finished_

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

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 Member Functions inherited from folly::Executor
virtual bool	keepAliveAcquire ()

virtual void	keepAliveRelease ()

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)

Detailed Description

Definition at line 55 of file ThreadedExecutor.h.

Constructor & Destructor Documentation

folly::ThreadedExecutor::ThreadedExecutor ( std::shared_ptr< ThreadFactory > threadFactory = newDefaultThreadFactory() )

explicit

Definition at line 34 of file ThreadedExecutor.cpp.

References control(), and controlt_.

     : threadFactory_(std::move(threadFactory)) {
   controlt_ = std::thread([this] { control(); });
 }

folly::ThreadedExecutor::~ThreadedExecutor ( )

override

Definition at line 39 of file ThreadedExecutor.cpp.

References controlt_, finished_, notify(), running_, and stopping_.

                                     {
   stopping_.store(true, std::memory_order_release);
   notify();
   controlt_.join();
   CHECK(running_.empty());
   CHECK(finished_.empty());
 }

folly::ThreadedExecutor::ThreadedExecutor ( ThreadedExecutor const & )

delete

folly::ThreadedExecutor::ThreadedExecutor ( ThreadedExecutor && )

delete

Member Function Documentation

void folly::ThreadedExecutor::add ( Func )

overridevirtual

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

Implements folly::Executor.

Definition at line 47 of file ThreadedExecutor.cpp.

References enqueued_, enqueuedm_, folly::gen::move, notify(), stopping_, and folly::with_unique_lock().

                                     {
   CHECK(!stopping_.load(std::memory_order_acquire));
   with_unique_lock(enqueuedm_, [&] { enqueued_.push_back(std::move(func)); });
   notify();
 }

void folly::ThreadedExecutor::control ( )

private

Definition at line 62 of file ThreadedExecutor.cpp.

References controlPerformAll(), controlWait(), and folly::setThreadName().

Referenced by ThreadedExecutor().

                                {
   folly::setThreadName("ThreadedCtrl");
   auto looping = true;
   while (looping) {
     controlWait();
     looping = controlPerformAll();
   }
 }

void folly::ThreadedExecutor::controlJoinFinishedThreads ( )

private

Definition at line 85 of file ThreadedExecutor.cpp.

References finished_, finishedm_, running_, folly::f14::swap(), and folly::with_unique_lock().

Referenced by controlPerformAll().

                                                   {
   std::deque<std::thread::id> finishedt;
   with_unique_lock(finishedm_, [&] { std::swap(finishedt, finished_); });
   for (auto id : finishedt) {
     running_[id].join();
     running_.erase(id);
   }
 }

void folly::ThreadedExecutor::controlLaunchEnqueuedTasks ( )

private

Definition at line 94 of file ThreadedExecutor.cpp.

References enqueued_, enqueuedm_, f, folly::gen::move, running_, folly::f14::swap(), threadFactory_, folly::with_unique_lock(), and work().

Referenced by controlPerformAll().

                                                   {
   std::deque<Func> enqueuedt;
   with_unique_lock(enqueuedm_, [&] { std::swap(enqueuedt, enqueued_); });
   for (auto& f : enqueuedt) {
     auto th = threadFactory_->newThread(
         [this, f = std::move(f)]() mutable { work(f); });
     auto id = th.get_id();
     running_[id] = std::move(th);
   }
 }

bool folly::ThreadedExecutor::controlPerformAll ( )

private

Definition at line 105 of file ThreadedExecutor.cpp.

References controlJoinFinishedThreads(), controlLaunchEnqueuedTasks(), running_, and stopping_.

Referenced by control().

                                          {
   auto stopping = stopping_.load(std::memory_order_acquire);
   controlJoinFinishedThreads();
   controlLaunchEnqueuedTasks();
   return !stopping || !running_.empty();
 }

void folly::ThreadedExecutor::controlWait ( )

private

Definition at line 71 of file ThreadedExecutor.cpp.

References controlc_, controlm_, controls_, and folly::lock().

Referenced by control().

                                    {
   constexpr auto kMaxWait = std::chrono::seconds(10);
   std::unique_lock<std::mutex> lock(controlm_);
   controlc_.wait_for(lock, kMaxWait, [&] { return controls_; });
   controls_ = false;
 }

std::shared_ptr< ThreadFactory > folly::ThreadedExecutor::newDefaultThreadFactory ( )

staticprivate

Definition at line 53 of file ThreadedExecutor.cpp.

                                                                        {
   return std::make_shared<NamedThreadFactory>("Threaded");
 }

void folly::ThreadedExecutor::notify ( )

private

Definition at line 57 of file ThreadedExecutor.cpp.

References controlc_, controlm_, controls_, and folly::with_unique_lock().

Referenced by add(), work(), and ~ThreadedExecutor().

                               {
   with_unique_lock(controlm_, [&] { controls_ = true; });
   controlc_.notify_one();
 }

ThreadedExecutor& folly::ThreadedExecutor::operator= ( ThreadedExecutor const & )

delete

ThreadedExecutor& folly::ThreadedExecutor::operator= ( ThreadedExecutor && )

delete

void folly::ThreadedExecutor::work ( Func & func )

private

Definition at line 78 of file ThreadedExecutor.cpp.

References finished_, finishedm_, notify(), and folly::with_unique_lock().

Referenced by controlLaunchEnqueuedTasks().

                                       {
   func();
   auto id = std::this_thread::get_id();
   with_unique_lock(finishedm_, [&] { finished_.push_back(id); });
   notify();
 }