Commit 7e5f29b8 authored by Alex Chow's avatar Alex Chow Committed by Facebook Github Bot

Add global ThreadPoolExecutor list, support naming ThreadPoolExecutors

Summary: There isn't a good way to access ThreadPoolExecutors to add instrumentation, such as grabbing basic stats on all executors to better understand pool capacity. This provides a way to do that.

Reviewed By: yfeldblum, djwatson

Differential Revision: D7658144

fbshipit-source-id: 3e72acdccf25742b76d438d2895d586049a0cc17
parent 6f9d8a14
...@@ -20,6 +20,14 @@ ...@@ -20,6 +20,14 @@
namespace folly { namespace folly {
using SyncVecThreadPoolExecutors =
folly::Synchronized<std::vector<ThreadPoolExecutor*>>;
SyncVecThreadPoolExecutors& getSyncVecThreadPoolExecutors() {
static Indestructible<SyncVecThreadPoolExecutors> storage;
return *storage;
}
ThreadPoolExecutor::ThreadPoolExecutor( ThreadPoolExecutor::ThreadPoolExecutor(
size_t /* numThreads */, size_t /* numThreads */,
std::shared_ptr<ThreadFactory> threadFactory, std::shared_ptr<ThreadFactory> threadFactory,
...@@ -27,10 +35,15 @@ ThreadPoolExecutor::ThreadPoolExecutor( ...@@ -27,10 +35,15 @@ ThreadPoolExecutor::ThreadPoolExecutor(
: threadFactory_(std::move(threadFactory)), : threadFactory_(std::move(threadFactory)),
isWaitForAll_(isWaitForAll), isWaitForAll_(isWaitForAll),
taskStatsCallbacks_(std::make_shared<TaskStatsCallbackRegistry>()), taskStatsCallbacks_(std::make_shared<TaskStatsCallbackRegistry>()),
threadPoolHook_("Wangle::ThreadPoolExecutor") {} threadPoolHook_("folly::ThreadPoolExecutor") {
getSyncVecThreadPoolExecutors()->push_back(this);
}
ThreadPoolExecutor::~ThreadPoolExecutor() { ThreadPoolExecutor::~ThreadPoolExecutor() {
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(
...@@ -173,6 +186,14 @@ void ThreadPoolExecutor::join() { ...@@ -173,6 +186,14 @@ void ThreadPoolExecutor::join() {
CHECK_EQ(0, stoppedThreads_.size()); CHECK_EQ(0, stoppedThreads_.size());
} }
void ThreadPoolExecutor::withAll(FunctionRef<void(ThreadPoolExecutor&)> f) {
getSyncVecThreadPoolExecutors().withRLock([f](auto& tpes) {
for (auto tpe : tpes) {
f(*tpe);
}
});
}
ThreadPoolExecutor::PoolStats ThreadPoolExecutor::getPoolStats() { ThreadPoolExecutor::PoolStats ThreadPoolExecutor::getPoolStats() {
const auto now = std::chrono::steady_clock::now(); const auto now = std::chrono::steady_clock::now();
RWSpinLock::ReadHolder r{&threadListLock_}; RWSpinLock::ReadHolder r{&threadListLock_};
...@@ -197,6 +218,15 @@ uint64_t ThreadPoolExecutor::getPendingTaskCount() { ...@@ -197,6 +218,15 @@ uint64_t ThreadPoolExecutor::getPendingTaskCount() {
return getPendingTaskCountImpl(); return getPendingTaskCountImpl();
} }
std::string ThreadPoolExecutor::getName() {
auto ntf = dynamic_cast<NamedThreadFactory*>(threadFactory_.get());
if (ntf == nullptr) {
return folly::demangle(typeid(*this).name()).toStdString();
}
return ntf->getNamePrefix();
}
std::atomic<uint64_t> ThreadPoolExecutor::Thread::nextId(0); std::atomic<uint64_t> ThreadPoolExecutor::Thread::nextId(0);
void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) { void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) {
......
...@@ -63,6 +63,12 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -63,6 +63,12 @@ class ThreadPoolExecutor : public virtual folly::Executor {
void stop(); void stop();
void join(); void join();
/**
* Execute f against all ThreadPoolExecutors, primarily for retrieving and
* exporting stats.
*/
static void withAll(FunctionRef<void(ThreadPoolExecutor&)> f);
struct PoolStats { struct PoolStats {
PoolStats() PoolStats()
: threadCount(0), : threadCount(0),
...@@ -78,6 +84,7 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -78,6 +84,7 @@ class ThreadPoolExecutor : public virtual folly::Executor {
PoolStats getPoolStats(); PoolStats getPoolStats();
uint64_t getPendingTaskCount(); uint64_t getPendingTaskCount();
std::string getName();
struct TaskStats { struct TaskStats {
TaskStats() : expired(false), waitTime(0), runTime(0) {} TaskStats() : expired(false), waitTime(0), runTime(0) {}
......
...@@ -651,3 +651,47 @@ TEST(ThreadPoolExecutorTest, KeepAliveTestIO) { ...@@ -651,3 +651,47 @@ TEST(ThreadPoolExecutorTest, KeepAliveTestIO) {
TEST(ThreadPoolExecutorTest, KeepAliveTestCPU) { TEST(ThreadPoolExecutorTest, KeepAliveTestCPU) {
keepAliveTest<CPUThreadPoolExecutor>(); keepAliveTest<CPUThreadPoolExecutor>();
} }
int getNumThreadPoolExecutors() {
int count = 0;
ThreadPoolExecutor::withAll([&count](ThreadPoolExecutor&) { count++; });
return count;
}
template <typename TPE>
static void registersToExecutorListTest() {
EXPECT_EQ(0, getNumThreadPoolExecutors());
{
TPE tpe(10);
EXPECT_EQ(1, getNumThreadPoolExecutors());
{
TPE tpe2(5);
EXPECT_EQ(2, getNumThreadPoolExecutors());
}
EXPECT_EQ(1, getNumThreadPoolExecutors());
}
EXPECT_EQ(0, getNumThreadPoolExecutors());
}
TEST(ThreadPoolExecutorTest, registersToExecutorListTestIO) {
registersToExecutorListTest<IOThreadPoolExecutor>();
}
TEST(ThreadPoolExecutorTest, registersToExecutorListTestCPU) {
registersToExecutorListTest<CPUThreadPoolExecutor>();
}
template <typename TPE>
static void testUsesNameFromNamedThreadFactory() {
auto ntf = std::make_shared<NamedThreadFactory>("my_executor");
TPE tpe(10, ntf);
EXPECT_EQ("my_executor", tpe.getName());
}
TEST(ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryIO) {
testUsesNameFromNamedThreadFactory<IOThreadPoolExecutor>();
}
TEST(ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryCPU) {
testUsesNameFromNamedThreadFactory<CPUThreadPoolExecutor>();
}
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