Skip to content

F
folly
Commit ea6ee538 authored by Lee Howes's avatar Lee Howes Committed by Facebook Github Bot
Browse files
Options

Support inline defer 4/n - Use variant of KeepAlive<> and DeferredExecutor in Core 

Summary:
After this diff, DeferredExecutor participates consistently in executor inline behaviour by being special cased in the core.

DeferredExecutor is no longer an executor, and is hence no longer special cased in the Future code. This is replaced with a variant of DeferredExecutor and Executor in Core.

Reviewed By: yfeldblum, andriigrynenko

Differential Revision: D15836529

fbshipit-source-id: 8324ba1de57e85fc757ecc3b431bf71858868a0d
parent 0c0aeccc
Hide whitespace changes
Inline Side-by-side
Showing with 406 additions and 154 deletions
folly/futures/Future-inl.h
View file @ ea6ee538
...@@ -379,7 +379,7 @@ FutureBase<T>::thenImplementation( ...@@ -379,7 +379,7 @@ FutureBase<T>::thenImplementation(
// grab the Future now before we lose our handle on the Promise // grab the Future now before we lose our handle on the Promise
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
sf.setExecutor(this->getExecutor()); sf.setExecutor(folly::Executor::KeepAlive<>{this->getExecutor()});
auto f = Future<B>(sf.core_); auto f = Future<B>(sf.core_);
sf.core_ = nullptr; sf.core_ = nullptr;
...@@ -726,26 +726,8 @@ SemiFuture<T> SemiFuture<T>::makeEmpty() { ...@@ -726,26 +726,8 @@ SemiFuture<T> SemiFuture<T>::makeEmpty() {
} }
template <class T> template <class T>
typename SemiFuture<T>::DeferredExecutor* SemiFuture<T>::getDeferredExecutor() futures::detail::DeferredWrapper SemiFuture<T>::stealDeferredExecutor() {
const { return this->getCore().stealDeferredExecutor();
if (auto executor = this->getExecutor()) {
assert(dynamic_cast<DeferredExecutor*>(executor) != nullptr);
return static_cast<DeferredExecutor*>(executor);
}
return nullptr;
}
template <class T>
folly::Executor::KeepAlive<typename SemiFuture<T>::DeferredExecutor>
SemiFuture<T>::stealDeferredExecutor() const {
if (auto executor = this->getExecutor()) {
assert(dynamic_cast<DeferredExecutor*>(executor) != nullptr);
auto executorKeepAlive =
folly::getKeepAliveToken(static_cast<DeferredExecutor*>(executor));
this->core_->setExecutor(nullptr);
return executorKeepAlive;
}
return {};
} }
template <class T> template <class T>
...@@ -753,10 +735,8 @@ void SemiFuture<T>::releaseDeferredExecutor(Core* core) { ...@@ -753,10 +735,8 @@ void SemiFuture<T>::releaseDeferredExecutor(Core* core) {
if (!core || core->hasCallback()) { if (!core || core->hasCallback()) {
return; return;
} }
if (auto executor = core->getExecutor()) { if (auto executor = core->stealDeferredExecutor()) {
assert(dynamic_cast<DeferredExecutor*>(executor) != nullptr); executor.get()->detach();
static_cast<DeferredExecutor*>(executor)->detach();
core->setExecutor(nullptr);
} }
} }
...@@ -774,7 +754,7 @@ SemiFuture<T>::SemiFuture(Future<T>&& other) noexcept ...@@ -774,7 +754,7 @@ SemiFuture<T>::SemiFuture(Future<T>&& other) noexcept
: futures::detail::FutureBase<T>(std::move(other)) { : futures::detail::FutureBase<T>(std::move(other)) {
// SemiFuture should not have an executor on construction // SemiFuture should not have an executor on construction
if (this->core_) { if (this->core_) {
this->setExecutor(nullptr); this->setExecutor(futures::detail::KeepAliveOrDeferred{});
} }
} }
...@@ -791,7 +771,7 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept { ...@@ -791,7 +771,7 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept {
this->assign(std::move(other)); this->assign(std::move(other));
// SemiFuture should not have an executor on construction // SemiFuture should not have an executor on construction
if (this->core_) { if (this->core_) {
this->setExecutor(nullptr); this->setExecutor(Executor::KeepAlive<>{});
} }
return *this; return *this;
} }
...@@ -802,7 +782,7 @@ Future<T> SemiFuture<T>::via(Executor::KeepAlive<> executor) && { ...@@ -802,7 +782,7 @@ Future<T> SemiFuture<T>::via(Executor::KeepAlive<> executor) && {
throw_exception<FutureNoExecutor>(); throw_exception<FutureNoExecutor>();
} }
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = this->getDeferredExecutor()) {
deferredExecutor->setExecutor(executor.copy()); deferredExecutor->setExecutor(executor.copy());
} }
...@@ -830,18 +810,24 @@ template <class T> ...@@ -830,18 +810,24 @@ template <class T>
template <typename F> template <typename F>
SemiFuture<typename futures::detail::tryCallableResult<T, F>::value_type> SemiFuture<typename futures::detail::tryCallableResult<T, F>::value_type>
SemiFuture<T>::defer(F&& func) && { SemiFuture<T>::defer(F&& func) && {
DeferredExecutor* deferredExecutor = getDeferredExecutor(); auto deferredExecutorPtr = this->getDeferredExecutor();
if (!deferredExecutor) { futures::detail::KeepAliveOrDeferred deferredExecutor = [&]() {
auto newDeferredExecutor = DeferredExecutor::create(); if (deferredExecutorPtr) {
deferredExecutor = newDeferredExecutor.get(); return futures::detail::KeepAliveOrDeferred{
this->setExecutor(std::move(newDeferredExecutor)); futures::detail::DeferredWrapper{deferredExecutorPtr}};
} } else {
auto newDeferredExecutor = futures::detail::KeepAliveOrDeferred(
futures::detail::DeferredWrapper::create());
this->setExecutor(newDeferredExecutor.copy());
return newDeferredExecutor;
}
}();
auto sf = Future<T>(this->core_).thenTry(std::forward<F>(func)).semi(); auto sf = Future<T>(this->core_).thenTry(std::forward<F>(func)).semi();
this->core_ = nullptr; this->core_ = nullptr;
// Carry deferred executor through chain as constructor from Future will // Carry deferred executor through chain as constructor from Future will
// nullify it // nullify it
sf.setExecutor(deferredExecutor); sf.setExecutor(std::move(deferredExecutor));
return sf; return sf;
} }
...@@ -850,35 +836,31 @@ template <typename F> ...@@ -850,35 +836,31 @@ template <typename F>
SemiFuture< SemiFuture<
typename futures::detail::tryExecutorCallableResult<T, F>::value_type> typename futures::detail::tryExecutorCallableResult<T, F>::value_type>
SemiFuture<T>::deferExTry(F&& func) && { SemiFuture<T>::deferExTry(F&& func) && {
DeferredExecutor* deferredExecutor = getDeferredExecutor(); auto deferredExecutorPtr = this->getDeferredExecutor();
if (!deferredExecutor) { futures::detail::DeferredWrapper deferredExecutor = [&]() mutable {
auto newDeferredExecutor = DeferredExecutor::create(); if (deferredExecutorPtr) {
deferredExecutor = newDeferredExecutor.get(); return futures::detail::DeferredWrapper(deferredExecutorPtr);
this->setExecutor(std::move(newDeferredExecutor)); } else {
} auto newDeferredExecutor = futures::detail::DeferredWrapper::create();
this->setExecutor(
auto sf = futures::detail::KeepAliveOrDeferred{newDeferredExecutor});
Future<T>(this->core_) return newDeferredExecutor;
.thenExTry([func = std::forward<F>(func)]( }
folly::Executor::KeepAlive<>&& keepAlive, }();
folly::Try<T>&& val) mutable {
// Extract the raw executor from the deferred and pass to the auto sf = Future<T>(this->core_)
// continuation .thenExTry([func = std::forward<F>(func)](
Executor* thenExDeferredExecutor = keepAlive.get(); folly::Executor::KeepAlive<>&& keepAlive,
assert( folly::Try<T>&& val) mutable {
dynamic_cast<DeferredExecutor*>(thenExDeferredExecutor) != return std::forward<F>(func)(
nullptr); std::move(keepAlive), std::forward<decltype(val)>(val));
auto innerExecutorKA = getKeepAliveToken( })
static_cast<DeferredExecutor*>(thenExDeferredExecutor) .semi();
->getExecutor());
return std::forward<F>(func)(
std::move(innerExecutorKA), std::forward<decltype(val)>(val));
})
.semi();
this->core_ = nullptr; this->core_ = nullptr;
// Carry deferred executor through chain as constructor from Future will // Carry deferred executor through chain as constructor from Future will
// nullify it // nullify it
sf.setExecutor(deferredExecutor); sf.setExecutor(
futures::detail::KeepAliveOrDeferred{std::move(deferredExecutor)});
return sf; return sf;
} }
...@@ -1432,24 +1414,23 @@ FOLLY_ALWAYS_INLINE FOLLY_ATTR_VISIBILITY_HIDDEN void foreach( ...@@ -1432,24 +1414,23 @@ FOLLY_ALWAYS_INLINE FOLLY_ATTR_VISIBILITY_HIDDEN void foreach(
} }
template <typename T> template <typename T>
DeferredExecutor* getDeferredExecutor(SemiFuture<T>& future) { futures::detail::DeferredExecutor* getDeferredExecutor(SemiFuture<T>& future) {
return future.getDeferredExecutor(); return future.getDeferredExecutor();
} }
template <typename T> template <typename T>
folly::Executor::KeepAlive<DeferredExecutor> stealDeferredExecutor( futures::detail::DeferredWrapper stealDeferredExecutor(SemiFuture<T>& future) {
SemiFuture<T>& future) {
return future.stealDeferredExecutor(); return future.stealDeferredExecutor();
} }
template <typename T> template <typename T>
folly::Executor::KeepAlive<DeferredExecutor> stealDeferredExecutor(Future<T>&) { futures::detail::DeferredWrapper stealDeferredExecutor(Future<T>&) {
return {}; return {};
} }
template <typename... Ts> template <typename... Ts>
void stealDeferredExecutorsVariadic( void stealDeferredExecutorsVariadic(
std::vector<folly::Executor::KeepAlive<DeferredExecutor>>& executors, std::vector<futures::detail::DeferredWrapper>& executors,
Ts&... ts) { Ts&... ts) {
auto foreach = [&](auto& future) { auto foreach = [&](auto& future) {
if (auto executor = stealDeferredExecutor(future)) { if (auto executor = stealDeferredExecutor(future)) {
...@@ -1462,7 +1443,7 @@ void stealDeferredExecutorsVariadic( ...@@ -1462,7 +1443,7 @@ void stealDeferredExecutorsVariadic(
template <class InputIterator> template <class InputIterator>
void stealDeferredExecutors( void stealDeferredExecutors(
std::vector<folly::Executor::KeepAlive<DeferredExecutor>>& executors, std::vector<futures::detail::DeferredWrapper>& executors,
InputIterator first, InputIterator first,
InputIterator last) { InputIterator last) {
for (auto it = first; it != last; ++it) { for (auto it = first; it != last; ++it) {
...@@ -1488,8 +1469,7 @@ collectAllSemiFuture(Fs&&... fs) { ...@@ -1488,8 +1469,7 @@ collectAllSemiFuture(Fs&&... fs) {
Result results; Result results;
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutorsVariadic(executors, fs...); futures::detail::stealDeferredExecutorsVariadic(executors, fs...);
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
...@@ -1537,8 +1517,7 @@ collectAllSemiFuture(InputIterator first, InputIterator last) { ...@@ -1537,8 +1517,7 @@ collectAllSemiFuture(InputIterator first, InputIterator last) {
std::vector<Try<T>> results; std::vector<Try<T>> results;
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutors(executors, first, last); futures::detail::stealDeferredExecutors(executors, first, last);
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last))); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
...@@ -1599,8 +1578,7 @@ collectSemiFuture(InputIterator first, InputIterator last) { ...@@ -1599,8 +1578,7 @@ collectSemiFuture(InputIterator first, InputIterator last) {
std::atomic<bool> threw{false}; std::atomic<bool> threw{false};
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutors(executors, first, last); futures::detail::stealDeferredExecutors(executors, first, last);
auto ctx = std::make_shared<Context>(std::distance(first, last)); auto ctx = std::make_shared<Context>(std::distance(first, last));
...@@ -1622,7 +1600,7 @@ collectSemiFuture(InputIterator first, InputIterator last) { ...@@ -1622,7 +1600,7 @@ collectSemiFuture(InputIterator first, InputIterator last) {
return std::move(t).value(); return std::move(t).value();
}; };
future = std::move(future).defer(work); future = std::move(future).defer(work);
auto deferredExecutor = futures::detail::getDeferredExecutor(future); const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors)); deferredExecutor->setNestedExecutors(std::move(executors));
} }
return future; return future;
...@@ -1653,8 +1631,7 @@ collectSemiFuture(Fs&&... fs) { ...@@ -1653,8 +1631,7 @@ collectSemiFuture(Fs&&... fs) {
std::atomic<bool> threw{false}; std::atomic<bool> threw{false};
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutorsVariadic(executors, fs...); futures::detail::stealDeferredExecutorsVariadic(executors, fs...);
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
...@@ -1678,7 +1655,7 @@ collectSemiFuture(Fs&&... fs) { ...@@ -1678,7 +1655,7 @@ collectSemiFuture(Fs&&... fs) {
return std::move(t).value(); return std::move(t).value();
}; };
future = std::move(future).defer(work); future = std::move(future).defer(work);
auto deferredExecutor = futures::detail::getDeferredExecutor(future); const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors)); deferredExecutor->setNestedExecutors(std::move(executors));
} }
return future; return future;
...@@ -1720,8 +1697,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) { ...@@ -1720,8 +1697,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) {
std::atomic<bool> done{false}; std::atomic<bool> done{false};
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutors(executors, first, last); futures::detail::stealDeferredExecutors(executors, first, last);
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
...@@ -1738,7 +1714,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) { ...@@ -1738,7 +1714,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) {
[](Try<typename decltype(future)::value_type>&& t) { [](Try<typename decltype(future)::value_type>&& t) {
return std::move(t).value(); return std::move(t).value();
}); });
auto deferredExecutor = futures::detail::getDeferredExecutor(future); const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors)); deferredExecutor->setNestedExecutors(std::move(executors));
} }
return future; return future;
...@@ -1762,8 +1738,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) { ...@@ -1762,8 +1738,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) {
size_t nTotal; size_t nTotal;
}; };
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutors(executors, first, last); futures::detail::stealDeferredExecutors(executors, first, last);
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last))); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
...@@ -1786,7 +1761,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) { ...@@ -1786,7 +1761,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) {
[](Try<typename decltype(future)::value_type>&& t) { [](Try<typename decltype(future)::value_type>&& t) {
return std::move(t).value(); return std::move(t).value();
}); });
auto deferredExecutor = futures::detail::getDeferredExecutor(future); const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors)); deferredExecutor->setNestedExecutors(std::move(executors));
} }
return future; return future;
...@@ -1822,8 +1797,7 @@ collectN(InputIterator first, InputIterator last, size_t n) { ...@@ -1822,8 +1797,7 @@ collectN(InputIterator first, InputIterator last, size_t n) {
exception_wrapper(std::runtime_error("Not enough futures"))); exception_wrapper(std::runtime_error("Not enough futures")));
} }
std::vector<folly::Executor::KeepAlive<futures::detail::DeferredExecutor>> std::vector<futures::detail::DeferredWrapper> executors;
executors;
futures::detail::stealDeferredExecutors(executors, first, last); futures::detail::stealDeferredExecutors(executors, first, last);
// for each completed Future, increase count and add to vector, until we // for each completed Future, increase count and add to vector, until we
...@@ -1863,7 +1837,7 @@ collectN(InputIterator first, InputIterator last, size_t n) { ...@@ -1863,7 +1837,7 @@ collectN(InputIterator first, InputIterator last, size_t n) {
[](Try<typename decltype(future)::value_type>&& t) { [](Try<typename decltype(future)::value_type>&& t) {
return std::move(t).value(); return std::move(t).value();
}); });
auto deferredExecutor = futures::detail::getDeferredExecutor(future); const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors)); deferredExecutor->setNestedExecutors(std::move(executors));
} }
return future; return future;
...@@ -2221,7 +2195,7 @@ void waitViaImpl( ...@@ -2221,7 +2195,7 @@ void waitViaImpl(
template <class T> template <class T>
SemiFuture<T>& SemiFuture<T>::wait() & { SemiFuture<T>& SemiFuture<T>::wait() & {
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = this->getDeferredExecutor()) {
// Make sure that the last callback in the future chain will be run on the // Make sure that the last callback in the future chain will be run on the
// WaitExecutor. // WaitExecutor.
Promise<T> promise; Promise<T> promise;
...@@ -2251,7 +2225,7 @@ SemiFuture<T>&& SemiFuture<T>::wait() && { ...@@ -2251,7 +2225,7 @@ SemiFuture<T>&& SemiFuture<T>::wait() && {
template <class T> template <class T>
SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & { SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & {
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = this->getDeferredExecutor()) {
// Make sure that the last callback in the future chain will be run on the // Make sure that the last callback in the future chain will be run on the
// WaitExecutor. // WaitExecutor.
Promise<T> promise; Promise<T> promise;
......
folly/futures/Future.h
View file @ ea6ee538
...@@ -407,11 +407,15 @@ class FutureBase { ...@@ -407,11 +407,15 @@ class FutureBase {
return getCore().getExecutor(); return getCore().getExecutor();
} }
DeferredExecutor* getDeferredExecutor() const {
return getCore().getDeferredExecutor();
}
// Sets the Executor within the Core state object of `this`. // Sets the Executor within the Core state object of `this`.
// Must be called either before attaching a callback or after the callback // Must be called either before attaching a callback or after the callback
// has already been invoked, but not concurrently with anything which might // has already been invoked, but not concurrently with anything which might
// trigger invocation of the callback. // trigger invocation of the callback.
void setExecutor(Executor::KeepAlive<> x) { void setExecutor(futures::detail::KeepAliveOrDeferred x) {
getCore().setExecutor(std::move(x)); getCore().setExecutor(std::move(x));
} }
...@@ -439,8 +443,7 @@ template <typename T> ...@@ -439,8 +443,7 @@ template <typename T>
DeferredExecutor* getDeferredExecutor(SemiFuture<T>& future); DeferredExecutor* getDeferredExecutor(SemiFuture<T>& future);
template <typename T> template <typename T>
folly::Executor::KeepAlive<DeferredExecutor> stealDeferredExecutor( futures::detail::DeferredWrapper stealDeferredExecutor(SemiFuture<T>& future);
SemiFuture<T>& future);
} // namespace detail } // namespace detail
} // namespace futures } // namespace futures
...@@ -795,8 +798,9 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -795,8 +798,9 @@ class SemiFuture : private futures::detail::FutureBase<T> {
if (deferredExecutor) { if (deferredExecutor) {
ret = ret =
std::move(ret).defer([](Try<T>&& t) { return std::move(t).value(); }); std::move(ret).defer([](Try<T>&& t) { return std::move(t).value(); });
ret.getDeferredExecutor()->setNestedExecutors( std::vector<futures::detail::DeferredWrapper> des;
{std::move(deferredExecutor)}); des.push_back(std::move(deferredExecutor));
ret.getDeferredExecutor()->setNestedExecutors(std::move(des));
} }
return ret; return ret;
} }
...@@ -852,9 +856,10 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -852,9 +856,10 @@ class SemiFuture : private futures::detail::FutureBase<T> {
friend class SemiFuture; friend class SemiFuture;
template <class> template <class>
friend class Future; friend class Future;
friend folly::Executor::KeepAlive<DeferredExecutor> friend futures::detail::DeferredWrapper
futures::detail::stealDeferredExecutor<T>(SemiFuture&); futures::detail::stealDeferredExecutor<T>(SemiFuture<T>&);
friend DeferredExecutor* futures::detail::getDeferredExecutor<T>(SemiFuture&); friend DeferredExecutor* futures::detail::getDeferredExecutor<T>(
SemiFuture<T>&);
using Base::setExecutor; using Base::setExecutor;
using Base::throwIfInvalid; using Base::throwIfInvalid;
...@@ -869,10 +874,7 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -869,10 +874,7 @@ class SemiFuture : private futures::detail::FutureBase<T> {
: Base(futures::detail::EmptyConstruct{}) {} : Base(futures::detail::EmptyConstruct{}) {}
// Throws FutureInvalid if !this->core_ // Throws FutureInvalid if !this->core_
DeferredExecutor* getDeferredExecutor() const; futures::detail::DeferredWrapper stealDeferredExecutor();
// Throws FutureInvalid if !this->core_
folly::Executor::KeepAlive<DeferredExecutor> stealDeferredExecutor() const;
/// Blocks until the future is fulfilled, or `dur` elapses. /// Blocks until the future is fulfilled, or `dur` elapses.
/// ///
......
folly/futures/detail/Core.h
View file @ ea6ee538
...@@ -86,24 +86,213 @@ bool compare_exchange_strong_release_acquire( ...@@ -86,24 +86,213 @@ bool compare_exchange_strong_release_acquire(
expected, desired, std::memory_order_release, std::memory_order_acquire); expected, desired, std::memory_order_release, std::memory_order_acquire);
} }
class DeferredExecutor;
/** /**
* Defer work until executor is actively boosted. * KeepAlive equivalent for DeferredExecutor, that is not a true executor.
* * KeepAliveOrDeferred acts as a union of DeferredWrapper and KeepAlive.
* NOTE: that this executor is a private implementation detail belonging to the */
* Folly Futures library and not intended to be used elsewhere. It is designed class DeferredWrapper {
* specifically for the use case of deferring work on a SemiFuture. It is NOT public:
* thread safe. Please do not use for any other purpose without great care. DeferredWrapper() = default;
/**
* Constructs a DeferredWrapper by acquiring the executor.
*/
explicit DeferredWrapper(DeferredExecutor* de);
DeferredWrapper(const DeferredWrapper& other);
DeferredWrapper(DeferredWrapper&& other)
: storage_{std::exchange(
other.storage_,
reinterpret_cast<uint64_t>(nullptr))} {}
DeferredWrapper& operator=(const DeferredWrapper& other);
DeferredWrapper& operator=(DeferredWrapper&& other) {
storage_ =
std::exchange(other.storage_, reinterpret_cast<uint64_t>(nullptr));
return *this;
}
DeferredExecutor* steal() {
uint64_t oldStorage =
std::exchange(storage_, reinterpret_cast<uint64_t>(nullptr));
return reinterpret_cast<DeferredExecutor*>(oldStorage & ~kDeferredFlag);
}
DeferredExecutor* get() const {
return reinterpret_cast<DeferredExecutor*>(storage_ & ~kDeferredFlag);
}
~DeferredWrapper();
explicit operator bool() const {
return storage_;
}
/**
* Check the passed value against the deferred flag and return true if it
* represents a DeferredWrapper.
*/
static bool representsDeferred(const uint64_t& storage) {
return (storage & kDeferredFlag) != 0;
}
static DeferredWrapper create();
// Bit to represent a deferred executor rather than a KeepAlive.
// Deferred bit will be bit 3 on 64-bit platforms, where pointers align to 8
// bytes so bit 2 is free. On 32-bit platforms pointers may align to 4 bytes
// so we have to use the rest of the uint64_t. On 32-bit big endian
// platforms where the pointer will sit at the high end of the uint64_t,
// also use bit 2. On little-endian 32-bit platforms use a bit above the
// size of the 32-bit pointer.
static constexpr uint64_t kDeferredFlag = uint64_t(1)
<< (std::numeric_limits<uintptr_t>::digits >= 64
? 2
: (std::numeric_limits<uintptr_t>::digits + 2));
static_assert(
!(std::numeric_limits<uintptr_t>::digits == 32 && kIsBigEndian),
"Big-endian 32-bit systems untested for Futures code.");
static_assert(
sizeof(uint64_t) >= sizeof(uintptr_t),
"Sanity check on pointer size");
static_assert(
(static_cast<uint64_t>(folly::detail::ExecutorKeepAliveBase::kFlagMask) &
kDeferredFlag) == 0,
"Nothing in the current executor mask can use the deferred bit.");
private:
uint64_t storage_ = reinterpret_cast<uint64_t>(nullptr);
};
/**
* Wrapper type that represents either a KeepAlive or a DeferredExecutor.
* Acts as if a type-safe tagged union of the two using knowledge that the two
* can safely be distinguished.
*/ */
class DeferredExecutor final : public Executor { class KeepAliveOrDeferred {
public: public:
void add(Func func) override { KeepAliveOrDeferred(Executor::KeepAlive<> ka) {
addFrom( new (&storage_) Executor::KeepAlive<>(std::move(ka));
Executor::KeepAlive<>{}, // Verify that the deferred bit is not set. If it were, KeepAlive and
[func = std::move(func)](Executor::KeepAlive<>&& /*ka*/) mutable { // DeferredExecutor would be impossible to distinguish.
func(); DCHECK(!isDeferred());
}); }
KeepAliveOrDeferred(DeferredWrapper deferred) {
new (&storage_) DeferredWrapper(std::move(deferred));
}
KeepAliveOrDeferred() {}
~KeepAliveOrDeferred() {
reset();
}
KeepAliveOrDeferred(KeepAliveOrDeferred&& other) {
if (other.isDeferred()) {
new (&storage_) DeferredWrapper(std::move(other).stealDeferred());
} else {
new (&storage_) Executor::KeepAlive<>(std::move(other).stealKeepAlive());
}
}
KeepAliveOrDeferred& operator=(KeepAliveOrDeferred&& other) {
reset();
if (other.isDeferred()) {
new (&storage_) DeferredWrapper(std::move(other).stealDeferred());
} else {
new (&storage_) Executor::KeepAlive<>(std::move(other).stealKeepAlive());
}
return *this;
}
DeferredExecutor* getDeferredExecutor() const {
if (!isDeferred()) {
return nullptr;
}
return asDeferred().get();
}
Executor* getKeepAliveExecutor() const {
if (isDeferred()) {
return nullptr;
}
return asKeepAlive().get();
}
Executor::KeepAlive<> stealKeepAlive() && {
if (isDeferred()) {
return Executor::KeepAlive<>{};
}
return std::move(asKeepAlive());
}
DeferredWrapper stealDeferred() && {
if (!isDeferred()) {
return DeferredWrapper{};
}
return std::move(asDeferred());
}
bool isDeferred() const {
return DeferredWrapper::representsDeferred(storage_);
}
bool isKeepAlive() const {
return !isDeferred();
}
KeepAliveOrDeferred copy() const {
if (isDeferred()) {
return KeepAliveOrDeferred{std::move(asDeferred())};
} else {
return KeepAliveOrDeferred{std::move(asKeepAlive())};
}
} }
explicit operator bool() const {
return storage_;
}
private:
uint64_t storage_ = reinterpret_cast<uintptr_t>(nullptr);
friend class DeferredExecutor;
void reset() {
if (isDeferred()) {
DeferredWrapper dw = std::move(asDeferred());
} else {
Executor::KeepAlive<> ka = std::move(asKeepAlive());
}
}
Executor::KeepAlive<>& asKeepAlive() {
return *reinterpret_cast<Executor::KeepAlive<>*>(&storage_);
}
const Executor::KeepAlive<>& asKeepAlive() const {
return *reinterpret_cast<const Executor::KeepAlive<>*>(&storage_);
}
DeferredWrapper& asDeferred() {
return *reinterpret_cast<DeferredWrapper*>(&storage_);
}
const DeferredWrapper& asDeferred() const {
return *reinterpret_cast<const DeferredWrapper*>(&storage_);
}
};
/**
* Defer work until executor is actively boosted.
*/
class DeferredExecutor final {
public:
// addFrom will: // addFrom will:
// * run func inline if there is a stored executor and completingKA matches // * run func inline if there is a stored executor and completingKA matches
// the stored executor // the stored executor
...@@ -158,7 +347,8 @@ class DeferredExecutor final : public Executor { ...@@ -158,7 +347,8 @@ class DeferredExecutor final : public Executor {
if (nestedExecutors_) { if (nestedExecutors_) {
auto nestedExecutors = std::exchange(nestedExecutors_, nullptr); auto nestedExecutors = std::exchange(nestedExecutors_, nullptr);
for (auto& nestedExecutor : *nestedExecutors) { for (auto& nestedExecutor : *nestedExecutors) {
nestedExecutor->setExecutor(executor.copy()); assert(nestedExecutor.get());
nestedExecutor.get()->setExecutor(executor.copy());
} }
} }
executor_ = std::move(executor); executor_ = std::move(executor);
...@@ -177,11 +367,18 @@ class DeferredExecutor final : public Executor { ...@@ -177,11 +367,18 @@ class DeferredExecutor final : public Executor {
executor_.copy().add(std::exchange(func_, nullptr)); executor_.copy().add(std::exchange(func_, nullptr));
} }
void setNestedExecutors(std::vector<DeferredWrapper> executors) {
DCHECK(!nestedExecutors_);
nestedExecutors_ =
std::make_unique<std::vector<DeferredWrapper>>(std::move(executors));
}
void detach() { void detach() {
if (nestedExecutors_) { if (nestedExecutors_) {
auto nestedExecutors = std::exchange(nestedExecutors_, nullptr); auto nestedExecutors = std::exchange(nestedExecutors_, nullptr);
for (auto& nestedExecutor : *nestedExecutors) { for (auto& nestedExecutor : *nestedExecutors) {
nestedExecutor->detach(); assert(nestedExecutor.get());
nestedExecutor.get()->detach();
} }
} }
auto state = state_.load(std::memory_order_acquire); auto state = state_.load(std::memory_order_acquire);
...@@ -199,29 +396,17 @@ class DeferredExecutor final : public Executor { ...@@ -199,29 +396,17 @@ class DeferredExecutor final : public Executor {
std::exchange(func_, nullptr); std::exchange(func_, nullptr);
} }
void setNestedExecutors(
std::vector<folly::Executor::KeepAlive<DeferredExecutor>> executors) {
DCHECK(!nestedExecutors_);
nestedExecutors_ = std::make_unique<
std::vector<folly::Executor::KeepAlive<DeferredExecutor>>>(
std::move(executors));
}
static KeepAlive<DeferredExecutor> create() {
return makeKeepAlive<DeferredExecutor>(new DeferredExecutor());
}
private: private:
DeferredExecutor() {} DeferredExecutor() {}
bool keepAliveAcquire() override { bool acquire() {
auto keepAliveCount = auto keepAliveCount =
keepAliveCount_.fetch_add(1, std::memory_order_relaxed); keepAliveCount_.fetch_add(1, std::memory_order_relaxed);
DCHECK(keepAliveCount > 0); DCHECK(keepAliveCount > 0);
return true; return true;
} }
void keepAliveRelease() override { void release() {
auto keepAliveCount = auto keepAliveCount =
keepAliveCount_.fetch_sub(1, std::memory_order_acq_rel); keepAliveCount_.fetch_sub(1, std::memory_order_acq_rel);
DCHECK(keepAliveCount > 0); DCHECK(keepAliveCount > 0);
...@@ -234,11 +419,48 @@ class DeferredExecutor final : public Executor { ...@@ -234,11 +419,48 @@ class DeferredExecutor final : public Executor {
std::atomic<State> state_{State::EMPTY}; std::atomic<State> state_{State::EMPTY};
Executor::KeepAlive<>::KeepAliveFunc func_; Executor::KeepAlive<>::KeepAliveFunc func_;
folly::Executor::KeepAlive<> executor_; folly::Executor::KeepAlive<> executor_;
std::unique_ptr<std::vector<folly::Executor::KeepAlive<DeferredExecutor>>> std::unique_ptr<std::vector<DeferredWrapper>> nestedExecutors_;
nestedExecutors_;
std::atomic<ssize_t> keepAliveCount_{1}; std::atomic<ssize_t> keepAliveCount_{1};
friend class KeepAliveOrDeferred;
friend class DeferredWrapper;
}; };
inline DeferredWrapper::DeferredWrapper(DeferredExecutor* de)
: storage_{reinterpret_cast<uint64_t>(de) | kDeferredFlag} {
de->acquire();
}
inline DeferredWrapper::~DeferredWrapper() {
if (storage_) {
steal()->release();
}
}
inline DeferredWrapper::DeferredWrapper(const DeferredWrapper& other)
: storage_{other.storage_} {
if (storage_) {
get()->acquire();
}
}
inline DeferredWrapper& DeferredWrapper::operator=(
const DeferredWrapper& other) {
storage_ = other.storage_;
if (storage_) {
get()->acquire();
}
return *this;
}
/* static */
inline DeferredWrapper DeferredWrapper::create() {
DeferredWrapper dw{};
auto* de = new DeferredExecutor{};
dw.storage_ = reinterpret_cast<uint64_t>(de) | kDeferredFlag;
return dw;
}
/// The shared state object for Future and Promise. /// The shared state object for Future and Promise.
/// ///
/// Nomenclature: /// Nomenclature:
...@@ -619,7 +841,7 @@ class Core final { ...@@ -619,7 +841,7 @@ class Core final {
/// Call only from consumer thread, either before attaching a callback or /// Call only from consumer thread, either before attaching a callback or
/// after the callback has already been invoked, but not concurrently with /// after the callback has already been invoked, but not concurrently with
/// anything which might trigger invocation of the callback. /// anything which might trigger invocation of the callback.
void setExecutor(Executor::KeepAlive<> x) { void setExecutor(KeepAliveOrDeferred&& x) {
DCHECK( DCHECK(
state_ != State::OnlyCallback && state_ != State::OnlyCallback &&
state_ != State::OnlyCallbackAllowInline); state_ != State::OnlyCallbackAllowInline);
...@@ -627,7 +849,26 @@ class Core final { ...@@ -627,7 +849,26 @@ class Core final {
} }
Executor* getExecutor() const { Executor* getExecutor() const {
return executor_.get(); if (!executor_.isKeepAlive()) {
return nullptr;
}
return executor_.getKeepAliveExecutor();
}
DeferredExecutor* getDeferredExecutor() const {
if (!executor_.isDeferred()) {
return {};
}
return executor_.getDeferredExecutor();
}
DeferredWrapper stealDeferredExecutor() {
if (executor_.isKeepAlive()) {
return {};
}
return std::move(executor_).stealDeferred();
} }
/// Call only from consumer thread /// Call only from consumer thread
...@@ -759,13 +1000,32 @@ class Core final { ...@@ -759,13 +1000,32 @@ class Core final {
void doCallback(Executor::KeepAlive<>&& completingKA, State priorState) { void doCallback(Executor::KeepAlive<>&& completingKA, State priorState) {
DCHECK(state_ == State::Done); DCHECK(state_ == State::Done);
auto executor = std::exchange(executor_, Executor::KeepAlive<>()); auto executor = std::exchange(executor_, KeepAliveOrDeferred{});
bool allowInline =
(executor.get() == completingKA.get() && // Customise inline behaviour
priorState == State::OnlyCallbackAllowInline); // If addCompletingKA is non-null, then we are allowing inline execution
// If we have no executor or if we are allowing inline executor on a auto doAdd = [](Executor::KeepAlive<>&& addCompletingKA,
// matching executor, run inline. KeepAliveOrDeferred&& currentExecutor,
if (executor && !allowInline) { auto&& keepAliveFunc) mutable {
if (auto deferredExecutorPtr = currentExecutor.getDeferredExecutor()) {
deferredExecutorPtr->addFrom(
std::move(addCompletingKA), std::move(keepAliveFunc));
} else {
// If executors match call inline
auto currentKeepAlive = std::move(currentExecutor).stealKeepAlive();
if (addCompletingKA.get() == currentKeepAlive.get()) {
keepAliveFunc(std::move(currentKeepAlive));
} else {
std::move(currentKeepAlive).add(std::move(keepAliveFunc));
}
}
};
if (executor) {
// If we are not allowing inline, clear the completing KA to disallow
if (!(priorState == State::OnlyCallbackAllowInline)) {
completingKA = Executor::KeepAlive<>{};
}
exception_wrapper ew; exception_wrapper ew;
// We need to reset `callback_` after it was executed (which can happen // We need to reset `callback_` after it was executed (which can happen
// through the executor or, if `Executor::add` throws, below). The // through the executor or, if `Executor::add` throws, below). The
...@@ -781,13 +1041,16 @@ class Core final { ...@@ -781,13 +1041,16 @@ class Core final {
CoreAndCallbackReference guard_local_scope(this); CoreAndCallbackReference guard_local_scope(this);
CoreAndCallbackReference guard_lambda(this); CoreAndCallbackReference guard_lambda(this);
try { try {
std::move(executor).add([core_ref = std::move(guard_lambda)]( doAdd(
Executor::KeepAlive<>&& ka) mutable { std::move(completingKA),
auto cr = std::move(core_ref); std::move(executor),
Core* const core = cr.getCore(); [core_ref =
RequestContextScopeGuard rctx(std::move(core->context_)); std::move(guard_lambda)](Executor::KeepAlive<>&& ka) mutable {
core->callback_(std::move(ka), std::move(core->result_)); auto cr = std::move(core_ref);
}); Core* const core = cr.getCore();
RequestContextScopeGuard rctx(std::move(core->context_));
core->callback_(std::move(ka), std::move(core->result_));
});
} catch (const std::exception& e) { } catch (const std::exception& e) {
ew = exception_wrapper(std::current_exception(), e); ew = exception_wrapper(std::current_exception(), e);
} catch (...) { } catch (...) {
...@@ -806,7 +1069,7 @@ class Core final { ...@@ -806,7 +1069,7 @@ class Core final {
detachOne(); detachOne();
}; };
RequestContextScopeGuard rctx(std::move(context_)); RequestContextScopeGuard rctx(std::move(context_));
callback_(std::move(executor), std::move(result_)); callback_(std::move(completingKA), std::move(result_));
} }
} }
...@@ -858,14 +1121,14 @@ class Core final { ...@@ -858,14 +1121,14 @@ class Core final {
std::atomic<unsigned char> callbackReferences_{0}; std::atomic<unsigned char> callbackReferences_{0};
std::atomic<bool> interruptHandlerSet_{false}; std::atomic<bool> interruptHandlerSet_{false};
SpinLock interruptLock_; SpinLock interruptLock_;
Executor::KeepAlive<> executor_; KeepAliveOrDeferred executor_;
union { union {
Context context_; Context context_;
}; };
std::unique_ptr<exception_wrapper> interrupt_{}; std::unique_ptr<exception_wrapper> interrupt_{};
std::function<void(exception_wrapper const&)> interruptHandler_{nullptr}; std::function<void(exception_wrapper const&)> interruptHandler_{nullptr};
}; };
} // namespace detail } // namespace detail
} // namespace futures } // namespace futures
} // namespace folly } // namespace folly
folly/futures/test/SemiFutureTest.cpp
View file @ ea6ee538
...@@ -1247,9 +1247,10 @@ TEST(SemiFuture, deferredExecutorInlineTest) { ...@@ -1247,9 +1247,10 @@ TEST(SemiFuture, deferredExecutorInlineTest) {
auto manualExec1KA = getKeepAliveToken(manualExec1); auto manualExec1KA = getKeepAliveToken(manualExec1);
auto manualExec2 = ManualExecutor{}; auto manualExec2 = ManualExecutor{};
auto manualExec2KA = getKeepAliveToken(manualExec2); auto manualExec2KA = getKeepAliveToken(manualExec2);
auto de = futures::detail::DeferredExecutor::create(); auto dw = futures::detail::DeferredWrapper::create();
auto* de = dw.get();
de->setExecutor(manualExec1KA); de->setExecutor(manualExec1KA);
de->add([&]() { a = true; }); de->addFrom(Executor::KeepAlive<>{}, [&](auto&&) { a = true; });
EXPECT_FALSE(a); EXPECT_FALSE(a);
manualExec1.run(); manualExec1.run();
EXPECT_TRUE(a); EXPECT_TRUE(a);
......
folly/futures/test/ViaTest.cpp
View file @ ea6ee538
...@@ -249,7 +249,7 @@ TEST(Via, then2) { ...@@ -249,7 +249,7 @@ TEST(Via, then2) {
TEST(Via, allowInline) { TEST(Via, allowInline) {
ManualExecutor x1, x2; ManualExecutor x1, x2;
bool a = false, b = false, c = false, d = false, e = false, f = false, bool a = false, b = false, c = false, d = false, e = false, f = false,
g = false, h = false, i = false, j = false; g = false, h = false, i = false, j = false, k = false, l = false;
via(&x1) via(&x1)
.thenValue([&](auto&&) { a = true; }) .thenValue([&](auto&&) { a = true; })
.thenTryInline([&](auto&&) { b = true; }) .thenTryInline([&](auto&&) { b = true; })
...@@ -265,7 +265,11 @@ TEST(Via, allowInline) { ...@@ -265,7 +265,11 @@ TEST(Via, allowInline) {
h = true; h = true;
return via(&x1).thenValue([&](auto&&) { i = true; }); return via(&x1).thenValue([&](auto&&) { i = true; });
}) })
.thenValueInline([&](auto&&) { j = true; }); .thenValueInline([&](auto&&) { j = true; })
.semi()
.deferValue([&](auto&&) { k = true; })
.via(&x2)
.thenValueInline([&](auto&&) { l = true; });
EXPECT_FALSE(a); EXPECT_FALSE(a);
EXPECT_FALSE(b); EXPECT_FALSE(b);
...@@ -307,8 +311,16 @@ TEST(Via, allowInline) { ...@@ -307,8 +311,16 @@ TEST(Via, allowInline) {
EXPECT_TRUE(i); EXPECT_TRUE(i);
EXPECT_FALSE(j); EXPECT_FALSE(j);
// Deferred work is not inline so k will remain false
x2.run(); x2.run();
EXPECT_TRUE(j); EXPECT_TRUE(j);
EXPECT_FALSE(k);
// Deferred work is not inline, but subsequent inline work should be inlined
// consistently with deferred work.
x2.run();
EXPECT_TRUE(k);
EXPECT_TRUE(l);
} }
#ifndef __APPLE__ // TODO #7372389 #ifndef __APPLE__ // TODO #7372389
......
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