Commit 8c7a26ad authored by Rodolfo Granata's avatar Rodolfo Granata Committed by Facebook Github Bot

folly/executors: properly synchronize access to thread pool stats.

Summary:
ThreadPoolExecutor::getPoolStats depends on derived classes implementation.
Calling `getPoolStats` is specially useful from `ThreadPoolExecutor::withAll`.

This change moves registration of thread pool executor instances to derived classes
to avoid `ThreadPoolExecutor::withAll` racing with thread pool destruction while calling `getPoolStats`.

Reviewed By: yfeldblum

Differential Revision: D17965135

fbshipit-source-id: 637de55ed78b085dc86fffa0e0f4a66cd4cfcede
parent e4519f66
...@@ -40,6 +40,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -40,6 +40,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::move(taskQueue)) { taskQueue_(std::move(taskQueue)) {
setNumThreads(numThreads); setNumThreads(numThreads);
registerThreadPoolExecutor(this);
} }
CPUThreadPoolExecutor::CPUThreadPoolExecutor( CPUThreadPoolExecutor::CPUThreadPoolExecutor(
...@@ -52,6 +53,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -52,6 +53,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::move(taskQueue)) { taskQueue_(std::move(taskQueue)) {
setNumThreads(numThreads.first); setNumThreads(numThreads.first);
registerThreadPoolExecutor(this);
} }
CPUThreadPoolExecutor::CPUThreadPoolExecutor( CPUThreadPoolExecutor::CPUThreadPoolExecutor(
...@@ -98,6 +100,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -98,6 +100,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
std::move(threadFactory)) {} std::move(threadFactory)) {}
CPUThreadPoolExecutor::~CPUThreadPoolExecutor() { CPUThreadPoolExecutor::~CPUThreadPoolExecutor() {
deregisterThreadPoolExecutor(this);
stop(); stop();
CHECK(threadsToStop_ == 0); CHECK(threadsToStop_ == 0);
} }
......
...@@ -141,7 +141,7 @@ class CPUThreadPoolExecutor : public ThreadPoolExecutor { ...@@ -141,7 +141,7 @@ class CPUThreadPoolExecutor : public ThreadPoolExecutor {
private: private:
void threadRun(ThreadPtr thread) override; void threadRun(ThreadPtr thread) override;
void stopThreads(size_t n) override; void stopThreads(size_t n) override;
size_t getPendingTaskCountImpl() const override; size_t getPendingTaskCountImpl() const override final;
bool tryDecrToStop(); bool tryDecrToStop();
bool taskShouldStop(folly::Optional<CPUTask>&); bool taskShouldStop(folly::Optional<CPUTask>&);
......
...@@ -248,9 +248,11 @@ EDFThreadPoolExecutor::EDFThreadPoolExecutor( ...@@ -248,9 +248,11 @@ EDFThreadPoolExecutor::EDFThreadPoolExecutor(
: ThreadPoolExecutor(numThreads, numThreads, std::move(threadFactory)), : ThreadPoolExecutor(numThreads, numThreads, std::move(threadFactory)),
taskQueue_(std::make_unique<TaskQueue>()) { taskQueue_(std::make_unique<TaskQueue>()) {
setNumThreads(numThreads); setNumThreads(numThreads);
registerThreadPoolExecutor(this);
} }
EDFThreadPoolExecutor::~EDFThreadPoolExecutor() { EDFThreadPoolExecutor::~EDFThreadPoolExecutor() {
deregisterThreadPoolExecutor(this);
stop(); stop();
} }
......
...@@ -60,7 +60,7 @@ class EDFThreadPoolExecutor : public SoftRealTimeExecutor, ...@@ -60,7 +60,7 @@ class EDFThreadPoolExecutor : public SoftRealTimeExecutor,
protected: protected:
void threadRun(ThreadPtr thread) override; void threadRun(ThreadPtr thread) override;
void stopThreads(std::size_t numThreads) override; void stopThreads(std::size_t numThreads) override;
std::size_t getPendingTaskCountImpl() const override; std::size_t getPendingTaskCountImpl() const override final;
private: private:
bool shouldStop(); bool shouldStop();
......
...@@ -80,9 +80,11 @@ IOThreadPoolExecutor::IOThreadPoolExecutor( ...@@ -80,9 +80,11 @@ IOThreadPoolExecutor::IOThreadPoolExecutor(
nextThread_(0), nextThread_(0),
eventBaseManager_(ebm) { eventBaseManager_(ebm) {
setNumThreads(numThreads); setNumThreads(numThreads);
registerThreadPoolExecutor(this);
} }
IOThreadPoolExecutor::~IOThreadPoolExecutor() { IOThreadPoolExecutor::~IOThreadPoolExecutor() {
deregisterThreadPoolExecutor(this);
stop(); stop();
} }
......
...@@ -86,7 +86,7 @@ class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor { ...@@ -86,7 +86,7 @@ class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor {
std::shared_ptr<IOThread> pickThread(); std::shared_ptr<IOThread> pickThread();
void threadRun(ThreadPtr thread) override; void threadRun(ThreadPtr thread) override;
void stopThreads(size_t n) override; void stopThreads(size_t n) override;
size_t getPendingTaskCountImpl() const override; size_t getPendingTaskCountImpl() const override final;
std::atomic<size_t> nextThread_; std::atomic<size_t> nextThread_;
folly::ThreadLocal<std::shared_ptr<IOThread>> thisThread_; folly::ThreadLocal<std::shared_ptr<IOThread>> thisThread_;
......
...@@ -29,6 +29,16 @@ SyncVecThreadPoolExecutors& getSyncVecThreadPoolExecutors() { ...@@ -29,6 +29,16 @@ SyncVecThreadPoolExecutors& getSyncVecThreadPoolExecutors() {
return *storage; return *storage;
} }
void ThreadPoolExecutor::registerThreadPoolExecutor(ThreadPoolExecutor* tpe) {
getSyncVecThreadPoolExecutors().wlock()->push_back(tpe);
}
void ThreadPoolExecutor::deregisterThreadPoolExecutor(ThreadPoolExecutor* tpe) {
getSyncVecThreadPoolExecutors().withWLock([tpe](auto& tpes) {
tpes.erase(std::remove(tpes.begin(), tpes.end(), tpe), tpes.end());
});
}
DEFINE_int64( DEFINE_int64(
threadtimeout_ms, threadtimeout_ms,
60000, 60000,
...@@ -44,16 +54,11 @@ ThreadPoolExecutor::ThreadPoolExecutor( ...@@ -44,16 +54,11 @@ ThreadPoolExecutor::ThreadPoolExecutor(
taskStatsCallbacks_(std::make_shared<TaskStatsCallbackRegistry>()), taskStatsCallbacks_(std::make_shared<TaskStatsCallbackRegistry>()),
threadPoolHook_("folly::ThreadPoolExecutor"), threadPoolHook_("folly::ThreadPoolExecutor"),
minThreads_(minThreads), minThreads_(minThreads),
threadTimeout_(FLAGS_threadtimeout_ms) { threadTimeout_(FLAGS_threadtimeout_ms) {}
getSyncVecThreadPoolExecutors().wlock()->push_back(this);
}
ThreadPoolExecutor::~ThreadPoolExecutor() { ThreadPoolExecutor::~ThreadPoolExecutor() {
joinKeepAliveOnce(); joinKeepAliveOnce();
CHECK_EQ(0, threadList_.get().size()); CHECK_EQ(0, threadList_.get().size());
getSyncVecThreadPoolExecutors().withWLock([this](auto& tpe) {
tpe.erase(std::remove(tpe.begin(), tpe.end(), this), tpe.end());
});
} }
ThreadPoolExecutor::Task::Task( ThreadPoolExecutor::Task::Task(
......
...@@ -53,6 +53,15 @@ namespace folly { ...@@ -53,6 +53,15 @@ namespace folly {
* ensureJoined() is called on add(), such that we can join idle * ensureJoined() is called on add(), such that we can join idle
* threads that were destroyed (which can't be joined from * threads that were destroyed (which can't be joined from
* themselves). * themselves).
*
* Thread pool stats accounting:
*
* Derived classes must register instances to keep stats on all thread
* pools by calling registerThreadPoolExecutor(this) on constructions
* and deregisterThreadPoolExecutor(this) on destruction.
*
* Registration must be done wherever getPendingTaskCountImpl is implemented
* and getPendingTaskCountImpl should be marked 'final' to avoid data races.
*/ */
class ThreadPoolExecutor : public DefaultKeepAliveExecutor { class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
public: public:
...@@ -224,10 +233,11 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor { ...@@ -224,10 +233,11 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
return std::make_shared<Thread>(this); return std::make_shared<Thread>(this);
} }
static void registerThreadPoolExecutor(ThreadPoolExecutor* tpe);
static void deregisterThreadPoolExecutor(ThreadPoolExecutor* tpe);
// Prerequisite: threadListLock_ readlocked or writelocked // Prerequisite: threadListLock_ readlocked or writelocked
virtual size_t getPendingTaskCountImpl() const { virtual size_t getPendingTaskCountImpl() const = 0;
throw std::logic_error("getPendingTaskCountImpl not implemented");
}
class ThreadList { class ThreadList {
public: public:
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment