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

Implement KeepAlive mechanism for thread pool executors

Reviewed By: yfeldblum

Differential Revision: D7765404

fbshipit-source-id: 3f09d6806d0fb98cb59529adc6ea2d51fa7ccbbb
parent 1aaded41
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <future>
#include <glog/logging.h>
#include <folly/Executor.h>
#include <folly/synchronization/Baton.h>
namespace folly {
/// An Executor accepts units of work with add(), which should be
/// threadsafe.
class DefaultKeepAliveExecutor : public virtual Executor {
public:
DefaultKeepAliveExecutor() : Executor() {}
virtual ~DefaultKeepAliveExecutor() {
DCHECK(!keepAlive_);
}
protected:
void joinKeepAlive() {
DCHECK(keepAlive_);
keepAlive_.reset();
keepAliveReleaseBaton_.wait();
}
private:
bool keepAliveAcquire() override {
auto keepAliveCounter =
keepAliveCounter_.fetch_add(1, std::memory_order_relaxed);
// We should never increment from 0
DCHECK(keepAliveCounter > 0);
return true;
}
void keepAliveRelease() override {
auto keepAliveCounter = --keepAliveCounter_;
DCHECK(keepAliveCounter >= 0);
if (keepAliveCounter == 0) {
keepAliveReleaseBaton_.post();
}
}
std::atomic<ssize_t> keepAliveCounter_{1};
Baton<> keepAliveReleaseBaton_;
KeepAlive keepAlive_{makeKeepAlive()};
};
} // namespace folly
...@@ -68,6 +68,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -68,6 +68,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
std::move(threadFactory)) {} std::move(threadFactory)) {}
CPUThreadPoolExecutor::~CPUThreadPoolExecutor() { CPUThreadPoolExecutor::~CPUThreadPoolExecutor() {
joinKeepAlive();
stop(); stop();
CHECK(threadsToStop_ == 0); CHECK(threadsToStop_ == 0);
} }
......
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#pragma once #pragma once
#include <folly/DefaultKeepAliveExecutor.h>
#include <folly/executors/ThreadPoolExecutor.h> #include <folly/executors/ThreadPoolExecutor.h>
namespace folly { namespace folly {
...@@ -60,7 +61,8 @@ namespace folly { ...@@ -60,7 +61,8 @@ 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,6 +73,7 @@ IOThreadPoolExecutor::IOThreadPoolExecutor( ...@@ -73,6 +73,7 @@ IOThreadPoolExecutor::IOThreadPoolExecutor(
} }
IOThreadPoolExecutor::~IOThreadPoolExecutor() { IOThreadPoolExecutor::~IOThreadPoolExecutor() {
joinKeepAlive();
stop(); stop();
} }
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#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>
...@@ -49,7 +50,9 @@ namespace folly { ...@@ -49,7 +50,9 @@ 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, public IOExecutor { class IOThreadPoolExecutor : public ThreadPoolExecutor,
public IOExecutor,
public DefaultKeepAliveExecutor {
public: public:
explicit IOThreadPoolExecutor( explicit IOThreadPoolExecutor(
size_t numThreads, size_t numThreads,
......
...@@ -628,3 +628,26 @@ TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestIO) { ...@@ -628,3 +628,26 @@ TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestIO) {
TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestCPU) { TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestCPU) {
resizeThreadWhileExecutingTest<CPUThreadPoolExecutor>(); resizeThreadWhileExecutingTest<CPUThreadPoolExecutor>();
} }
template <typename TPE>
void keepAliveTest() {
auto executor = std::make_unique<TPE>(4);
auto f =
futures::sleep(std::chrono::milliseconds{100})
.via(executor.get())
.then([keepAlive = executor->getKeepAliveToken()] { return 42; });
executor.reset();
EXPECT_TRUE(f.isReady());
EXPECT_EQ(42, f.get());
}
TEST(ThreadPoolExecutorTest, KeepAliveTestIO) {
keepAliveTest<IOThreadPoolExecutor>();
}
TEST(ThreadPoolExecutorTest, KeepAliveTestCPU) {
keepAliveTest<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