Commit 619679d3 authored by Tristan Rice's avatar Tristan Rice Committed by Facebook Github Bot

Future.within: cancel timeout callbacks on success

Summary:
Currently the timekeeper callback set by the within method isn't cleaned up when the future resolves and always waits the full duration. This can cause OOM under high QPS and long timeout situations.

This resolves this by cancelling the callback.

Reviewed By: mpark

Differential Revision: D16835816

fbshipit-source-id: c400afc37dbc66447aaa8fd557a770d717d9ab02
parent 83bddb4d
...@@ -503,6 +503,12 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && { ...@@ -503,6 +503,12 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && {
Future<Unit> thisFuture; Future<Unit> thisFuture;
Promise<T> promise; Promise<T> promise;
std::atomic<bool> token{false}; std::atomic<bool> token{false};
// The baton is required since the continuation in thisFuture and
// afterFuture need to access each other.
// Use a spin lock baton since this should be fulfilled almost immediately.
Baton</*MayBlock=*/false> thisFutureSet;
Future<Unit> afterFuture;
}; };
std::shared_ptr<Timekeeper> tks; std::shared_ptr<Timekeeper> tks;
...@@ -517,14 +523,43 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && { ...@@ -517,14 +523,43 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && {
auto ctx = std::make_shared<Context>(std::move(e)); auto ctx = std::make_shared<Context>(std::move(e));
// Have time keeper use a weak ptr to hold ctx,
// so that ctx can be deallocated as soon as the future job finished.
ctx->afterFuture = tk->after(dur).thenTry(
[weakCtx = to_weak_ptr(ctx)](Try<Unit>&& t) mutable {
if (t.hasException() &&
t.exception().is_compatible_with<FutureCancellation>()) {
// This got cancelled by thisFuture so we can just return.
return;
}
auto lockedCtx = weakCtx.lock();
if (!lockedCtx) {
// ctx already released. "this" completed first, cancel "after"
return;
}
// "after" completed first, cancel "this"
lockedCtx->thisFutureSet.wait();
lockedCtx->thisFuture.raise(FutureTimeout());
if (!lockedCtx->token.exchange(true, std::memory_order_relaxed)) {
if (t.hasException()) {
lockedCtx->promise.setException(std::move(t.exception()));
} else {
lockedCtx->promise.setException(std::move(lockedCtx->exception));
}
}
});
auto f = [ctx](Executor::KeepAlive<>&&, Try<T>&& t) { auto f = [ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
if (!ctx->token.exchange(true, std::memory_order_relaxed)) { if (!ctx->token.exchange(true, std::memory_order_relaxed)) {
ctx->promise.setTry(std::move(t)); ctx->promise.setTry(std::move(t));
ctx->afterFuture.cancel();
} }
}; };
using R = futures::detail::tryExecutorCallableResult<T, decltype(f)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(f)>;
ctx->thisFuture = this->thenImplementation( ctx->thisFuture = this->thenImplementation(
std::move(f), R{}, futures::detail::InlineContinuation::forbid); std::move(f), R{}, futures::detail::InlineContinuation::forbid);
ctx->thisFutureSet.post();
// Properly propagate interrupt values through futures chained after within() // Properly propagate interrupt values through futures chained after within()
ctx->promise.setInterruptHandler( ctx->promise.setInterruptHandler(
...@@ -534,25 +569,6 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && { ...@@ -534,25 +569,6 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && {
} }
}); });
// Have time keeper use a weak ptr to hold ctx,
// so that ctx can be deallocated as soon as the future job finished.
tk->after(dur).thenTry([weakCtx = to_weak_ptr(ctx)](Try<Unit>&& t) mutable {
auto lockedCtx = weakCtx.lock();
if (!lockedCtx) {
// ctx already released. "this" completed first, cancel "after"
return;
}
// "after" completed first, cancel "this"
lockedCtx->thisFuture.raise(FutureTimeout());
if (!lockedCtx->token.exchange(true, std::memory_order_relaxed)) {
if (t.hasException()) {
lockedCtx->promise.setException(std::move(t.exception()));
} else {
lockedCtx->promise.setException(std::move(lockedCtx->exception));
}
}
});
return ctx->promise.getSemiFuture(); return ctx->promise.getSemiFuture();
} }
......
...@@ -116,6 +116,31 @@ TEST(Timekeeper, semiFutureWithinHandlesNullTimekeeperSingleton) { ...@@ -116,6 +116,31 @@ TEST(Timekeeper, semiFutureWithinHandlesNullTimekeeperSingleton) {
EXPECT_THROW(std::move(f).get(), FutureNoTimekeeper); EXPECT_THROW(std::move(f).get(), FutureNoTimekeeper);
} }
TEST(Timekeeper, semiFutureWithinCancelsTimeout) {
struct MockTimekeeper : Timekeeper {
MockTimekeeper() {
p_.setInterruptHandler([this](const exception_wrapper& ew) {
ew.handle([this](const FutureCancellation&) { cancelled_ = true; });
p_.setException(ew);
});
}
Future<Unit> after(Duration) override {
return p_.getFuture();
}
Promise<Unit> p_;
bool cancelled_{false};
};
MockTimekeeper tk;
Promise<int> p;
auto f = p.getSemiFuture().within(too_long, static_cast<Timekeeper*>(&tk));
p.setValue(1);
EXPECT_TRUE(tk.cancelled_);
}
TEST(Timekeeper, futureDelayed) { TEST(Timekeeper, futureDelayed) {
auto t1 = now(); auto t1 = now();
auto dur = makeFuture() auto dur = makeFuture()
......
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