Commit b619a10f authored by Lee Howes's avatar Lee Howes Committed by Facebook Github Bot

Support inline defer 1/n - Add and use r-value add on KeepAlive

Summary:
Allows adding directly to a KeepAlive if the KeepAlive is consumed in the process.

This allows simultaneous consumption and propagation of the KeepAlive into the callback, propagating information about the executor running the callback without the cost of reference counting the KeepAlive.

Reviewed By: andriigrynenko

Differential Revision: D15836528

fbshipit-source-id: 8f9e6f6ec47ad294391741d25fce68a85a429919
parent 51ad358e
...@@ -90,6 +90,8 @@ class Executor { ...@@ -90,6 +90,8 @@ class Executor {
class KeepAlive : pushmi::folly_pipeorigin, class KeepAlive : pushmi::folly_pipeorigin,
private detail::ExecutorKeepAliveBase { private detail::ExecutorKeepAliveBase {
public: public:
using KeepAliveFunc = Function<void(KeepAlive&&)>;
KeepAlive() = default; KeepAlive() = default;
~KeepAlive() { ~KeepAlive() {
...@@ -183,6 +185,17 @@ class Executor { ...@@ -183,6 +185,17 @@ class Executor {
return KeepAlive(storage_ | kAliasFlag); return KeepAlive(storage_ | kAliasFlag);
} }
template <class KAF>
void add(KAF&& f) && {
static_assert(
is_invocable<KAF, KeepAlive&&>::value,
"Parameter to add must be void(KeepAlive&&)>");
auto ex = get();
ex->add([ka = std::move(*this), f = std::forward<KAF>(f)]() mutable {
f(std::move(ka));
});
}
private: private:
friend class Executor; friend class Executor;
template <typename OtherExecutor> template <typename OtherExecutor>
......
...@@ -571,12 +571,14 @@ class DeferredExecutor final : public Executor { ...@@ -571,12 +571,14 @@ class DeferredExecutor final : public Executor {
if (state == State::DETACHED) { if (state == State::DETACHED) {
return; return;
} }
auto kaFunc = [func = std::move(func)](
Executor::KeepAlive<>&& /* ka */) mutable { func(); };
if (state == State::HAS_EXECUTOR) { if (state == State::HAS_EXECUTOR) {
executor_->add(std::move(func)); executor_.copy().add(std::move(kaFunc));
return; return;
} }
DCHECK(state == State::EMPTY); DCHECK(state == State::EMPTY);
func_ = std::move(func); func_ = std::move(kaFunc);
if (detail::compare_exchange_strong_release_acquire( if (detail::compare_exchange_strong_release_acquire(
state_, state, State::HAS_FUNCTION)) { state_, state, State::HAS_FUNCTION)) {
return; return;
...@@ -586,7 +588,7 @@ class DeferredExecutor final : public Executor { ...@@ -586,7 +588,7 @@ class DeferredExecutor final : public Executor {
std::exchange(func_, nullptr); std::exchange(func_, nullptr);
return; return;
} }
executor_->add(std::exchange(func_, nullptr)); executor_.copy().add(std::exchange(func_, nullptr));
} }
Executor* getExecutor() const { Executor* getExecutor() const {
...@@ -611,7 +613,7 @@ class DeferredExecutor final : public Executor { ...@@ -611,7 +613,7 @@ class DeferredExecutor final : public Executor {
DCHECK(state == State::HAS_FUNCTION); DCHECK(state == State::HAS_FUNCTION);
state_.store(State::HAS_EXECUTOR, std::memory_order_release); state_.store(State::HAS_EXECUTOR, std::memory_order_release);
executor_->add(std::exchange(func_, nullptr)); executor_.copy().add(std::exchange(func_, nullptr));
} }
void detach() { void detach() {
...@@ -666,7 +668,7 @@ class DeferredExecutor final : public Executor { ...@@ -666,7 +668,7 @@ class DeferredExecutor final : public Executor {
enum class State { EMPTY, HAS_FUNCTION, HAS_EXECUTOR, DETACHED }; enum class State { EMPTY, HAS_FUNCTION, HAS_EXECUTOR, DETACHED };
std::atomic<State> state_{State::EMPTY}; std::atomic<State> state_{State::EMPTY};
Func 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<folly::Executor::KeepAlive<DeferredExecutor>>>
nestedExecutors_; nestedExecutors_;
......
...@@ -628,13 +628,12 @@ class Core final { ...@@ -628,13 +628,12 @@ class Core final {
CoreAndCallbackReference guard_local_scope(this); CoreAndCallbackReference guard_local_scope(this);
CoreAndCallbackReference guard_lambda(this); CoreAndCallbackReference guard_lambda(this);
try { try {
auto xPtr = executor.get(); std::move(executor).add([core_ref = std::move(guard_lambda)](
xPtr->add([core_ref = std::move(guard_lambda), Executor::KeepAlive<>&& ka) mutable {
executor = std::move(executor)]() mutable {
auto cr = std::move(core_ref); auto cr = std::move(core_ref);
Core* const core = cr.getCore(); Core* const core = cr.getCore();
RequestContextScopeGuard rctx(std::move(core->context_)); RequestContextScopeGuard rctx(std::move(core->context_));
core->callback_(std::move(executor), std::move(core->result_)); 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);
......
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