Commit f8651741 authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot

Move CPU/IOThreadPoolExecutor keep-alive logic into ThreadPoolExecutor

Summary: This makes keep-alive join also happen if stop/join are called directly.

Reviewed By: yfeldblum

Differential Revision: D8060479

fbshipit-source-id: 1d1c0cdbcaac2e439aee9ad14f00fa7e4fda2339
parent 109f7655
...@@ -88,7 +88,6 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -88,7 +88,6 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
std::move(threadFactory)) {} std::move(threadFactory)) {}
CPUThreadPoolExecutor::~CPUThreadPoolExecutor() { CPUThreadPoolExecutor::~CPUThreadPoolExecutor() {
joinKeepAlive();
stop(); stop();
CHECK(threadsToStop_ == 0); CHECK(threadsToStop_ == 0);
} }
......
...@@ -16,7 +16,6 @@ ...@@ -16,7 +16,6 @@
#pragma once #pragma once
#include <folly/DefaultKeepAliveExecutor.h>
#include <folly/executors/ThreadPoolExecutor.h> #include <folly/executors/ThreadPoolExecutor.h>
DECLARE_bool(dynamic_cputhreadpoolexecutor); DECLARE_bool(dynamic_cputhreadpoolexecutor);
...@@ -63,8 +62,7 @@ namespace folly { ...@@ -63,8 +62,7 @@ namespace folly {
* priority tasks could still hog all the threads. (at last check pthreads * priority tasks could still hog all the threads. (at last check pthreads
* thread priorities didn't work very well). * thread priorities didn't work very well).
*/ */
class CPUThreadPoolExecutor : public ThreadPoolExecutor, class CPUThreadPoolExecutor : public ThreadPoolExecutor {
public DefaultKeepAliveExecutor {
public: public:
struct CPUTask; struct CPUTask;
......
...@@ -73,7 +73,6 @@ IOThreadPoolExecutor::IOThreadPoolExecutor( ...@@ -73,7 +73,6 @@ IOThreadPoolExecutor::IOThreadPoolExecutor(
} }
IOThreadPoolExecutor::~IOThreadPoolExecutor() { IOThreadPoolExecutor::~IOThreadPoolExecutor() {
joinKeepAlive();
stop(); stop();
} }
......
...@@ -18,7 +18,6 @@ ...@@ -18,7 +18,6 @@
#include <atomic> #include <atomic>
#include <folly/DefaultKeepAliveExecutor.h>
#include <folly/executors/IOExecutor.h> #include <folly/executors/IOExecutor.h>
#include <folly/executors/ThreadPoolExecutor.h> #include <folly/executors/ThreadPoolExecutor.h>
#include <folly/io/async/EventBaseManager.h> #include <folly/io/async/EventBaseManager.h>
...@@ -50,9 +49,7 @@ namespace folly { ...@@ -50,9 +49,7 @@ namespace folly {
* outstanding tasks belong to the event base and will be executed upon its * outstanding tasks belong to the event base and will be executed upon its
* destruction. * destruction.
*/ */
class IOThreadPoolExecutor : public ThreadPoolExecutor, class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor {
public IOExecutor,
public DefaultKeepAliveExecutor {
public: public:
explicit IOThreadPoolExecutor( explicit IOThreadPoolExecutor(
size_t numThreads, size_t numThreads,
......
...@@ -48,6 +48,7 @@ ThreadPoolExecutor::ThreadPoolExecutor( ...@@ -48,6 +48,7 @@ ThreadPoolExecutor::ThreadPoolExecutor(
} }
ThreadPoolExecutor::~ThreadPoolExecutor() { ThreadPoolExecutor::~ThreadPoolExecutor() {
joinKeepAliveOnce();
CHECK_EQ(0, threadList_.get().size()); CHECK_EQ(0, threadList_.get().size());
getSyncVecThreadPoolExecutors().withWLock([this](auto& tpe) { getSyncVecThreadPoolExecutors().withWLock([this](auto& tpe) {
tpe.erase(std::remove(tpe.begin(), tpe.end(), this), tpe.end()); tpe.erase(std::remove(tpe.begin(), tpe.end(), this), tpe.end());
...@@ -207,6 +208,7 @@ void ThreadPoolExecutor::joinStoppedThreads(size_t n) { ...@@ -207,6 +208,7 @@ void ThreadPoolExecutor::joinStoppedThreads(size_t n) {
} }
void ThreadPoolExecutor::stop() { void ThreadPoolExecutor::stop() {
joinKeepAliveOnce();
size_t n = 0; size_t n = 0;
{ {
SharedMutex::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
...@@ -223,6 +225,7 @@ void ThreadPoolExecutor::stop() { ...@@ -223,6 +225,7 @@ void ThreadPoolExecutor::stop() {
} }
void ThreadPoolExecutor::join() { void ThreadPoolExecutor::join() {
joinKeepAliveOnce();
size_t n = 0; size_t n = 0;
{ {
SharedMutex::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
......
...@@ -14,7 +14,7 @@ ...@@ -14,7 +14,7 @@
* limitations under the License. * limitations under the License.
*/ */
#pragma once #pragma once
#include <folly/Executor.h> #include <folly/DefaultKeepAliveExecutor.h>
#include <folly/Memory.h> #include <folly/Memory.h>
#include <folly/SharedMutex.h> #include <folly/SharedMutex.h>
#include <folly/executors/GlobalThreadPoolList.h> #include <folly/executors/GlobalThreadPoolList.h>
...@@ -52,7 +52,7 @@ namespace folly { ...@@ -52,7 +52,7 @@ namespace folly {
* threads that were destroyed (which can't be joined from * threads that were destroyed (which can't be joined from
* themselves). * themselves).
*/ */
class ThreadPoolExecutor : public virtual folly::Executor { class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
public: public:
explicit ThreadPoolExecutor( explicit ThreadPoolExecutor(
size_t maxThreads, size_t maxThreads,
...@@ -308,6 +308,14 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -308,6 +308,14 @@ class ThreadPoolExecutor : public virtual folly::Executor {
std::atomic<size_t> threadsToJoin_{0}; std::atomic<size_t> threadsToJoin_{0};
std::chrono::milliseconds threadTimeout_{0}; std::chrono::milliseconds threadTimeout_{0};
void joinKeepAliveOnce() {
if (!std::exchange(keepAliveJoined_, true)) {
joinKeepAlive();
}
}
bool keepAliveJoined_{false};
}; };
} // namespace folly } // namespace folly
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