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

Add keep-alive support to ManualExecutor

Reviewed By: lewissbaker

Differential Revision: D15363378

fbshipit-source-id: 0343d6cc42624a6c3e9b01a338afb1646625928a
parent 43e27ca3
...@@ -23,6 +23,9 @@ ...@@ -23,6 +23,9 @@
namespace folly { namespace folly {
ManualExecutor::~ManualExecutor() { ManualExecutor::~ManualExecutor() {
while (keepAliveCount_.load(std::memory_order_relaxed)) {
drive();
}
drain(); drain();
} }
...@@ -68,6 +71,7 @@ size_t ManualExecutor::run() { ...@@ -68,6 +71,7 @@ size_t ManualExecutor::run() {
funcs_.pop(); funcs_.pop();
} }
func(); func();
func = nullptr;
} }
return count; return count;
......
...@@ -124,6 +124,17 @@ class ManualExecutor : public DrivableExecutor, ...@@ -124,6 +124,17 @@ class ManualExecutor : public DrivableExecutor,
return funcs.size() + scheduled_funcs.size(); return funcs.size() + scheduled_funcs.size();
} }
bool keepAliveAcquire() override {
keepAliveCount_.fetch_add(1, std::memory_order_relaxed);
return true;
}
void keepAliveRelease() override {
if (keepAliveCount_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
add([] {});
}
}
private: private:
std::mutex lock_; std::mutex lock_;
std::queue<Func> funcs_; std::queue<Func> funcs_;
...@@ -156,6 +167,8 @@ class ManualExecutor : public DrivableExecutor, ...@@ -156,6 +167,8 @@ class ManualExecutor : public DrivableExecutor,
}; };
std::priority_queue<ScheduledFunc> scheduledFuncs_; std::priority_queue<ScheduledFunc> scheduledFuncs_;
TimePoint now_ = TimePoint::min(); TimePoint now_ = TimePoint::min();
std::atomic<ssize_t> keepAliveCount_{0};
}; };
} // namespace folly } // namespace folly
...@@ -210,6 +210,18 @@ TEST(ManualExecutor, drainsOnDestruction) { ...@@ -210,6 +210,18 @@ TEST(ManualExecutor, drainsOnDestruction) {
EXPECT_EQ(1, count); EXPECT_EQ(1, count);
} }
TEST(ManualExecutor, keepAlive) {
auto future = [] {
ManualExecutor ex;
return futures::sleep(std::chrono::milliseconds{100})
.via(&ex)
.thenValue([](auto) { return 42; })
.semi();
}();
EXPECT_TRUE(future.isReady());
EXPECT_EQ(42, std::move(future).get());
}
TEST(Executor, InlineExecutor) { TEST(Executor, InlineExecutor) {
InlineExecutor x; InlineExecutor x;
size_t counter = 0; size_t counter = 0;
......
...@@ -547,11 +547,14 @@ TEST(Via, viaExecutorDiscardsTaskFutureSetValueFirst) { ...@@ -547,11 +547,14 @@ TEST(Via, viaExecutorDiscardsTaskFutureSetValueFirst) {
Promise<Unit> captured_promise; Promise<Unit> captured_promise;
auto captured_promise_future = captured_promise.getFuture(); auto captured_promise_future = captured_promise.getFuture();
Optional<Future<int>> future; Optional<SemiFuture<int>> future;
{ {
ManualExecutor x; ManualExecutor x;
future = makeFuture().via(&x).thenValue( future =
[c = std::move(captured_promise)](auto&&) { return 42; }); makeFuture()
.via(&x)
.thenValue([c = std::move(captured_promise)](auto&&) { return 42; })
.semi();
x.clear(); x.clear();
} }
...@@ -570,12 +573,15 @@ TEST(Via, viaExecutorDiscardsTaskFutureSetCallbackFirst) { ...@@ -570,12 +573,15 @@ TEST(Via, viaExecutorDiscardsTaskFutureSetCallbackFirst) {
Promise<Unit> captured_promise; Promise<Unit> captured_promise;
auto captured_promise_future = captured_promise.getFuture(); auto captured_promise_future = captured_promise.getFuture();
Optional<Future<int>> future; Optional<SemiFuture<int>> future;
{ {
ManualExecutor x; ManualExecutor x;
Promise<Unit> trigger; Promise<Unit> trigger;
future = trigger.getFuture().via(&x).thenValue( future =
[c = std::move(captured_promise)](auto&&) { return 42; }); trigger.getFuture()
.via(&x)
.thenValue([c = std::move(captured_promise)](auto&&) { return 42; })
.semi();
trigger.setValue(); trigger.setValue();
x.clear(); x.clear();
} }
......
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