Commit 20665b48 authored by Dave Watson's avatar Dave Watson Committed by Facebook Github Bot

Change to folly::SharedMutex

Summary:
One (artificial) test case creates ~1000 threads, resulting in lots of spinning contending for this lock.

Changing to folly::SharedMutex fixes the issue, allowing the actual working thread to finish.

Reviewed By: magedm

Differential Revision: D7847705

fbshipit-source-id: b2a8c3acdb4c62f347ef4ec761acbe7681ced9f3
parent a478d52f
...@@ -185,7 +185,7 @@ void CPUThreadPoolExecutor::threadRun(ThreadPtr thread) { ...@@ -185,7 +185,7 @@ void CPUThreadPoolExecutor::threadRun(ThreadPtr thread) {
o->threadStopped(thread.get()); o->threadStopped(thread.get());
} }
// Actually remove the thread from the list. // Actually remove the thread from the list.
folly::RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
threadList_.remove(thread); threadList_.remove(thread);
stoppedThreads_.add(thread); stoppedThreads_.add(thread);
return; return;
...@@ -198,7 +198,7 @@ void CPUThreadPoolExecutor::threadRun(ThreadPtr thread) { ...@@ -198,7 +198,7 @@ void CPUThreadPoolExecutor::threadRun(ThreadPtr thread) {
if (UNLIKELY(threadsToStop_ > 0 && !isJoin_)) { if (UNLIKELY(threadsToStop_ > 0 && !isJoin_)) {
if (tryDecrToStop()) { if (tryDecrToStop()) {
folly::RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
threadList_.remove(thread); threadList_.remove(thread);
stoppedThreads_.add(thread); stoppedThreads_.add(thread);
return; return;
......
...@@ -86,7 +86,7 @@ void IOThreadPoolExecutor::add( ...@@ -86,7 +86,7 @@ void IOThreadPoolExecutor::add(
std::chrono::milliseconds expiration, std::chrono::milliseconds expiration,
Func expireCallback) { Func expireCallback) {
ensureActiveThreads(); ensureActiveThreads();
RWSpinLock::ReadHolder r{&threadListLock_}; SharedMutex::ReadHolder r{&threadListLock_};
if (threadList_.get().empty()) { if (threadList_.get().empty()) {
throw std::runtime_error("No threads available"); throw std::runtime_error("No threads available");
} }
...@@ -127,7 +127,7 @@ IOThreadPoolExecutor::pickThread() { ...@@ -127,7 +127,7 @@ IOThreadPoolExecutor::pickThread() {
EventBase* IOThreadPoolExecutor::getEventBase() { EventBase* IOThreadPoolExecutor::getEventBase() {
ensureActiveThreads(); ensureActiveThreads();
RWSpinLock::ReadHolder r{&threadListLock_}; SharedMutex::ReadHolder r{&threadListLock_};
return pickThread()->eventBase; return pickThread()->eventBase;
} }
......
...@@ -136,7 +136,7 @@ void ThreadPoolExecutor::setNumThreads(size_t numThreads) { ...@@ -136,7 +136,7 @@ void ThreadPoolExecutor::setNumThreads(size_t numThreads) {
size_t numThreadsToJoin = 0; size_t numThreadsToJoin = 0;
{ {
RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
auto pending = getPendingTaskCountImpl(); auto pending = getPendingTaskCountImpl();
maxThreads_.store(numThreads, std::memory_order_relaxed); maxThreads_.store(numThreads, std::memory_order_relaxed);
auto active = activeThreads_.load(std::memory_order_relaxed); auto active = activeThreads_.load(std::memory_order_relaxed);
...@@ -208,7 +208,7 @@ void ThreadPoolExecutor::joinStoppedThreads(size_t n) { ...@@ -208,7 +208,7 @@ void ThreadPoolExecutor::joinStoppedThreads(size_t n) {
void ThreadPoolExecutor::stop() { void ThreadPoolExecutor::stop() {
{ {
folly::RWSpinLock::WriteHolder w{&threadListLock_}; folly::SharedMutex::WriteHolder w{&threadListLock_};
maxThreads_.store(0, std::memory_order_release); maxThreads_.store(0, std::memory_order_release);
activeThreads_.store(0, std::memory_order_release); activeThreads_.store(0, std::memory_order_release);
} }
...@@ -216,7 +216,7 @@ void ThreadPoolExecutor::stop() { ...@@ -216,7 +216,7 @@ void ThreadPoolExecutor::stop() {
size_t n = 0; size_t n = 0;
{ {
RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
n = threadList_.get().size(); n = threadList_.get().size();
removeThreads(n, false); removeThreads(n, false);
} }
...@@ -227,7 +227,7 @@ void ThreadPoolExecutor::stop() { ...@@ -227,7 +227,7 @@ void ThreadPoolExecutor::stop() {
void ThreadPoolExecutor::join() { void ThreadPoolExecutor::join() {
{ {
folly::RWSpinLock::WriteHolder w{&threadListLock_}; folly::SharedMutex::WriteHolder w{&threadListLock_};
maxThreads_.store(0, std::memory_order_release); maxThreads_.store(0, std::memory_order_release);
activeThreads_.store(0, std::memory_order_release); activeThreads_.store(0, std::memory_order_release);
} }
...@@ -235,7 +235,7 @@ void ThreadPoolExecutor::join() { ...@@ -235,7 +235,7 @@ void ThreadPoolExecutor::join() {
size_t n = 0; size_t n = 0;
{ {
RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
n = threadList_.get().size(); n = threadList_.get().size();
removeThreads(n, true); removeThreads(n, true);
} }
...@@ -254,7 +254,7 @@ void ThreadPoolExecutor::withAll(FunctionRef<void(ThreadPoolExecutor&)> f) { ...@@ -254,7 +254,7 @@ void ThreadPoolExecutor::withAll(FunctionRef<void(ThreadPoolExecutor&)> f) {
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_}; SharedMutex::ReadHolder r{&threadListLock_};
ThreadPoolExecutor::PoolStats stats; ThreadPoolExecutor::PoolStats stats;
size_t activeTasks = 0; size_t activeTasks = 0;
size_t idleAlive = 0; size_t idleAlive = 0;
...@@ -278,7 +278,7 @@ ThreadPoolExecutor::PoolStats ThreadPoolExecutor::getPoolStats() { ...@@ -278,7 +278,7 @@ ThreadPoolExecutor::PoolStats ThreadPoolExecutor::getPoolStats() {
} }
size_t ThreadPoolExecutor::getPendingTaskCount() { size_t ThreadPoolExecutor::getPendingTaskCount() {
RWSpinLock::ReadHolder r{&threadListLock_}; SharedMutex::ReadHolder r{&threadListLock_};
return getPendingTaskCountImpl(); return getPendingTaskCountImpl();
} }
...@@ -346,7 +346,7 @@ size_t ThreadPoolExecutor::StoppedThreadQueue::size() { ...@@ -346,7 +346,7 @@ size_t ThreadPoolExecutor::StoppedThreadQueue::size() {
void ThreadPoolExecutor::addObserver(std::shared_ptr<Observer> o) { void ThreadPoolExecutor::addObserver(std::shared_ptr<Observer> o) {
{ {
RWSpinLock::ReadHolder r{&threadListLock_}; SharedMutex::ReadHolder r{&threadListLock_};
observers_.push_back(o); observers_.push_back(o);
for (auto& thread : threadList_.get()) { for (auto& thread : threadList_.get()) {
o->threadPreviouslyStarted(thread.get()); o->threadPreviouslyStarted(thread.get());
...@@ -359,7 +359,7 @@ void ThreadPoolExecutor::addObserver(std::shared_ptr<Observer> o) { ...@@ -359,7 +359,7 @@ void ThreadPoolExecutor::addObserver(std::shared_ptr<Observer> o) {
} }
void ThreadPoolExecutor::removeObserver(std::shared_ptr<Observer> o) { void ThreadPoolExecutor::removeObserver(std::shared_ptr<Observer> o) {
RWSpinLock::ReadHolder r{&threadListLock_}; SharedMutex::ReadHolder r{&threadListLock_};
for (auto& thread : threadList_.get()) { for (auto& thread : threadList_.get()) {
o->threadNotYetStopped(thread.get()); o->threadNotYetStopped(thread.get());
} }
...@@ -397,7 +397,7 @@ void ThreadPoolExecutor::ensureActiveThreads() { ...@@ -397,7 +397,7 @@ void ThreadPoolExecutor::ensureActiveThreads() {
return; return;
} }
RWSpinLock::WriteHolder w{&threadListLock_}; SharedMutex::WriteHolder w{&threadListLock_};
// Double check behind lock. // Double check behind lock.
active = activeThreads_.load(std::memory_order_relaxed); active = activeThreads_.load(std::memory_order_relaxed);
total = maxThreads_.load(std::memory_order_relaxed); total = maxThreads_.load(std::memory_order_relaxed);
......
...@@ -16,12 +16,12 @@ ...@@ -16,12 +16,12 @@
#pragma once #pragma once
#include <folly/Executor.h> #include <folly/Executor.h>
#include <folly/Memory.h> #include <folly/Memory.h>
#include <folly/SharedMutex.h>
#include <folly/executors/GlobalThreadPoolList.h> #include <folly/executors/GlobalThreadPoolList.h>
#include <folly/executors/task_queue/LifoSemMPMCQueue.h> #include <folly/executors/task_queue/LifoSemMPMCQueue.h>
#include <folly/executors/thread_factory/NamedThreadFactory.h> #include <folly/executors/thread_factory/NamedThreadFactory.h>
#include <folly/io/async/Request.h> #include <folly/io/async/Request.h>
#include <folly/synchronization/Baton.h> #include <folly/synchronization/Baton.h>
#include <folly/synchronization/RWSpinLock.h>
#include <algorithm> #include <algorithm>
#include <mutex> #include <mutex>
...@@ -282,7 +282,7 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -282,7 +282,7 @@ class ThreadPoolExecutor : public virtual folly::Executor {
const bool isWaitForAll_; // whether to wait till event base loop exits const bool isWaitForAll_; // whether to wait till event base loop exits
ThreadList threadList_; ThreadList threadList_;
folly::RWSpinLock threadListLock_; SharedMutex threadListLock_;
StoppedThreadQueue stoppedThreads_; StoppedThreadQueue stoppedThreads_;
std::atomic<bool> isJoin_{false}; // whether the current downsizing is a join std::atomic<bool> isJoin_{false}; // whether the current downsizing is a join
......
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