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

Remove priority from Core

Summary: Wrap Executor in ExecutorWithPriority if custom priority is passed.

Reviewed By: yfeldblum

Differential Revision: D14079222

fbshipit-source-id: 56917827f54115b90ca237c71321f7db327355e1
parent cd6352dd
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <thread> #include <thread>
#include <folly/Optional.h> #include <folly/Optional.h>
#include <folly/executors/ExecutorWithPriority.h>
#include <folly/executors/InlineExecutor.h> #include <folly/executors/InlineExecutor.h>
#include <folly/executors/QueuedImmediateExecutor.h> #include <folly/executors/QueuedImmediateExecutor.h>
#include <folly/futures/detail/Core.h> #include <folly/futures/detail/Core.h>
...@@ -853,9 +854,7 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept { ...@@ -853,9 +854,7 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept {
} }
template <class T> template <class T>
Future<T> SemiFuture<T>::via( Future<T> SemiFuture<T>::via(Executor::KeepAlive<> executor) && {
Executor::KeepAlive<> executor,
int8_t priority) && {
if (!executor) { if (!executor) {
throw_exception<FutureNoExecutor>(); throw_exception<FutureNoExecutor>();
} }
...@@ -866,11 +865,19 @@ Future<T> SemiFuture<T>::via( ...@@ -866,11 +865,19 @@ Future<T> SemiFuture<T>::via(
auto newFuture = Future<T>(this->core_); auto newFuture = Future<T>(this->core_);
this->core_ = nullptr; this->core_ = nullptr;
newFuture.setExecutor(std::move(executor), priority); newFuture.setExecutor(std::move(executor));
return newFuture; return newFuture;
} }
template <class T>
Future<T> SemiFuture<T>::via(
Executor::KeepAlive<> executor,
int8_t priority) && {
return std::move(*this).via(
ExecutorWithPriority::create(std::move(executor), priority));
}
template <class T> template <class T>
Future<T> SemiFuture<T>::toUnsafeFuture() && { Future<T> SemiFuture<T>::toUnsafeFuture() && {
return std::move(*this).via(&InlineExecutor::instance()); return std::move(*this).via(&InlineExecutor::instance());
...@@ -1029,8 +1036,8 @@ typename std:: ...@@ -1029,8 +1036,8 @@ typename std::
} }
template <class T> template <class T>
Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) && { Future<T> Future<T>::via(Executor::KeepAlive<> executor) && {
this->setExecutor(std::move(executor), priority); this->setExecutor(std::move(executor));
auto newFuture = Future<T>(this->core_); auto newFuture = Future<T>(this->core_);
this->core_ = nullptr; this->core_ = nullptr;
...@@ -1038,7 +1045,13 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) && { ...@@ -1038,7 +1045,13 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) && {
} }
template <class T> template <class T>
Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) & { Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) && {
return std::move(*this).via(
ExecutorWithPriority::create(std::move(executor), priority));
}
template <class T>
Future<T> Future<T>::via(Executor::KeepAlive<> executor) & {
this->throwIfInvalid(); this->throwIfInvalid();
Promise<T> p; Promise<T> p;
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
...@@ -1052,7 +1065,12 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) & { ...@@ -1052,7 +1065,12 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) & {
// check in SemiFuture::via // check in SemiFuture::via
auto f = Future<T>(sf.core_); auto f = Future<T>(sf.core_);
sf.core_ = nullptr; sf.core_ = nullptr;
return std::move(f).via(std::move(executor), priority); return std::move(f).via(std::move(executor));
}
template <class T>
Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) & {
return this->via(ExecutorWithPriority::create(std::move(executor), priority));
} }
template <typename T> template <typename T>
...@@ -1388,6 +1406,10 @@ Future<T> makeFuture(Try<T> t) { ...@@ -1388,6 +1406,10 @@ Future<T> makeFuture(Try<T> t) {
} }
// via // via
Future<Unit> via(Executor::KeepAlive<> executor) {
return makeFuture().via(std::move(executor));
}
Future<Unit> via(Executor::KeepAlive<> executor, int8_t priority) { Future<Unit> via(Executor::KeepAlive<> executor, int8_t priority) {
return makeFuture().via(std::move(executor), priority); return makeFuture().via(std::move(executor), priority);
} }
......
...@@ -344,11 +344,6 @@ class FutureBase { ...@@ -344,11 +344,6 @@ class FutureBase {
raise(FutureCancellation()); raise(FutureCancellation());
} }
// Returns this future's executor priority.
int8_t getPriority() const {
return getCore().getPriority();
}
protected: protected:
friend class Promise<T>; friend class Promise<T>;
template <class> template <class>
...@@ -417,10 +412,8 @@ class FutureBase { ...@@ -417,10 +412,8 @@ class FutureBase {
// 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( void setExecutor(Executor::KeepAlive<> x) {
Executor::KeepAlive<> x, getCore().setExecutor(std::move(x));
int8_t priority = Executor::MID_PRI) {
getCore().setExecutor(std::move(x), priority);
} }
// Variant: returns a value // Variant: returns a value
...@@ -551,7 +544,6 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -551,7 +544,6 @@ class SemiFuture : private futures::detail::FutureBase<T> {
/* implicit */ SemiFuture(Future<T>&&) noexcept; /* implicit */ SemiFuture(Future<T>&&) noexcept;
using Base::cancel; using Base::cancel;
using Base::getPriority;
using Base::hasException; using Base::hasException;
using Base::hasValue; using Base::hasValue;
using Base::isReady; using Base::isReady;
...@@ -665,9 +657,8 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -665,9 +657,8 @@ class SemiFuture : private futures::detail::FutureBase<T> {
bool wait(Duration dur) &&; bool wait(Duration dur) &&;
/// Returns a Future which will call back on the other side of executor. /// Returns a Future which will call back on the other side of executor.
Future<T> via( Future<T> via(Executor::KeepAlive<> executor) &&;
Executor::KeepAlive<> executor, Future<T> via(Executor::KeepAlive<> executor, int8_t priority) &&;
int8_t priority = Executor::MID_PRI) &&;
/// Defer work to run on the consumer of the future. /// Defer work to run on the consumer of the future.
/// Function must take a Try as a parameter. /// Function must take a Try as a parameter.
...@@ -1008,7 +999,6 @@ class Future : private futures::detail::FutureBase<T> { ...@@ -1008,7 +999,6 @@ class Future : private futures::detail::FutureBase<T> {
Future& operator=(Future<T2>&&); Future& operator=(Future<T2>&&);
using Base::cancel; using Base::cancel;
using Base::getPriority;
using Base::hasException; using Base::hasException;
using Base::hasValue; using Base::hasValue;
using Base::isReady; using Base::isReady;
...@@ -1082,9 +1072,8 @@ class Future : private futures::detail::FutureBase<T> { ...@@ -1082,9 +1072,8 @@ class Future : private futures::detail::FutureBase<T> {
/// ///
/// - `valid() == false` /// - `valid() == false`
/// - `RESULT.valid() == true` /// - `RESULT.valid() == true`
Future<T> via( Future<T> via(Executor::KeepAlive<> executor) &&;
Executor::KeepAlive<> executor, Future<T> via(Executor::KeepAlive<> executor, int8_t priority) &&;
int8_t priority = Executor::MID_PRI) &&;
/// Returns a Future which will call back on the other side of executor. /// Returns a Future which will call back on the other side of executor.
/// ///
...@@ -1099,9 +1088,8 @@ class Future : private futures::detail::FutureBase<T> { ...@@ -1099,9 +1088,8 @@ class Future : private futures::detail::FutureBase<T> {
/// - `valid() == true` /// - `valid() == true`
/// - `RESULT.valid() == true` /// - `RESULT.valid() == true`
/// - when `this` gets fulfilled, it automatically fulfills RESULT /// - when `this` gets fulfilled, it automatically fulfills RESULT
Future<T> via( Future<T> via(Executor::KeepAlive<> executor) &;
Executor::KeepAlive<> executor, Future<T> via(Executor::KeepAlive<> executor, int8_t priority) &;
int8_t priority = Executor::MID_PRI) &;
/// When this Future has completed, execute func which is a function that /// When this Future has completed, execute func which is a function that
/// can be called with either `T&&` or `Try<T>&&`. /// can be called with either `T&&` or `Try<T>&&`.
......
...@@ -51,8 +51,7 @@ class FutureSplitter { ...@@ -51,8 +51,7 @@ class FutureSplitter {
*/ */
explicit FutureSplitter(Future<T>&& future) explicit FutureSplitter(Future<T>&& future)
: promise_(std::make_shared<SharedPromise<T>>()), : promise_(std::make_shared<SharedPromise<T>>()),
e_(getExecutorFrom(future)), e_(getExecutorFrom(future)) {
priority_(future.getPriority()) {
std::move(future).thenTry([promise = promise_](Try<T>&& theTry) { std::move(future).thenTry([promise = promise_](Try<T>&& theTry) {
promise->setTry(std::move(theTry)); promise->setTry(std::move(theTry));
}); });
...@@ -65,7 +64,7 @@ class FutureSplitter { ...@@ -65,7 +64,7 @@ class FutureSplitter {
if (promise_ == nullptr) { if (promise_ == nullptr) {
throw_exception<FutureSplitterInvalid>(); throw_exception<FutureSplitterInvalid>();
} }
return promise_->getSemiFuture().via(e_, priority_); return promise_->getSemiFuture().via(e_);
} }
/** /**
...@@ -81,7 +80,6 @@ class FutureSplitter { ...@@ -81,7 +80,6 @@ class FutureSplitter {
private: private:
std::shared_ptr<SharedPromise<T>> promise_; std::shared_ptr<SharedPromise<T>> promise_;
Executor* e_ = nullptr; Executor* e_ = nullptr;
int8_t priority_{-1};
static Executor* getExecutorFrom(Future<T>& f) { static Executor* getExecutorFrom(Future<T>& f) {
// If the passed future had a null executor, use an inline executor // If the passed future had a null executor, use an inline executor
......
...@@ -421,26 +421,15 @@ class Core final { ...@@ -421,26 +421,15 @@ 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( void setExecutor(Executor::KeepAlive<> x) {
Executor::KeepAlive<> x,
int8_t priority = Executor::MID_PRI) {
DCHECK(state_ != State::OnlyCallback); DCHECK(state_ != State::OnlyCallback);
executor_ = std::move(x); executor_ = std::move(x);
priority_ = priority;
}
void setExecutor(Executor* x, int8_t priority = Executor::MID_PRI) {
setExecutor(getKeepAliveToken(x), priority);
} }
Executor* getExecutor() const { Executor* getExecutor() const {
return executor_.get(); return executor_.get();
} }
int8_t getPriority() const {
return priority_;
}
/// Call only from consumer thread /// Call only from consumer thread
/// ///
/// Eventual effect is to pass `e` to the Promise's interrupt handler, either /// Eventual effect is to pass `e` to the Promise's interrupt handler, either
...@@ -572,7 +561,6 @@ class Core final { ...@@ -572,7 +561,6 @@ class Core final {
if (executor_) { if (executor_) {
auto x = exchange(executor_, Executor::KeepAlive<>()); auto x = exchange(executor_, Executor::KeepAlive<>());
int8_t priority = priority_;
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
...@@ -590,25 +578,13 @@ class Core final { ...@@ -590,25 +578,13 @@ class Core final {
CoreAndCallbackReference guard_lambda(this); CoreAndCallbackReference guard_lambda(this);
try { try {
auto xPtr = x.get(); auto xPtr = x.get();
if (LIKELY(x->getNumPriorities() == 1)) { xPtr->add([core_ref = std::move(guard_lambda),
xPtr->add([core_ref = std::move(guard_lambda), keepAlive = std::move(x)]() mutable {
keepAlive = std::move(x)]() 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(core->context_);
RequestContextScopeGuard rctx(core->context_); core->callback_(std::move(core->result_));
core->callback_(std::move(core->result_)); });
});
} else {
xPtr->addWithPriority(
[core_ref = std::move(guard_lambda),
keepAlive = std::move(x)]() mutable {
auto cr = std::move(core_ref);
Core* const core = cr.getCore();
RequestContextScopeGuard rctx(core->context_);
core->callback_(std::move(core->result_));
},
priority);
}
} 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 (...) {
...@@ -633,7 +609,7 @@ class Core final { ...@@ -633,7 +609,7 @@ class Core final {
void proxyCallback() { void proxyCallback() {
state_.store(State::Empty, std::memory_order_relaxed); state_.store(State::Empty, std::memory_order_relaxed);
proxy_->setExecutor(std::move(executor_), priority_); proxy_->setExecutor(std::move(executor_));
proxy_->setCallback(std::move(callback_), std::move(context_)); proxy_->setCallback(std::move(callback_), std::move(context_));
proxy_->detachFuture(); proxy_->detachFuture();
context_.~Context(); context_.~Context();
...@@ -673,7 +649,6 @@ class Core final { ...@@ -673,7 +649,6 @@ 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_;
int8_t priority_{-1};
Executor::KeepAlive<> executor_; Executor::KeepAlive<> executor_;
union { union {
Context context_; Context context_;
......
...@@ -315,9 +315,8 @@ Future<T> makeFuture(Try<T> t); ...@@ -315,9 +315,8 @@ Future<T> makeFuture(Try<T> t);
* *
* @returns a void Future that will call back on the given executor * @returns a void Future that will call back on the given executor
*/ */
inline Future<Unit> via( inline Future<Unit> via(Executor::KeepAlive<> executor);
Executor::KeepAlive<> executor, inline Future<Unit> via(Executor::KeepAlive<> executor, int8_t priority);
int8_t priority = Executor::MID_PRI);
/// Execute a function via the given executor and return a future. /// Execute a function via the given executor and return a future.
/// This is semantically equivalent to via(executor).then(func), but /// This is semantically equivalent to via(executor).then(func), but
......
...@@ -174,19 +174,3 @@ TEST(FutureSplitter, splitFutureFailure) { ...@@ -174,19 +174,3 @@ TEST(FutureSplitter, splitFutureFailure) {
EXPECT_TRUE(f2.isReady()); EXPECT_TRUE(f2.isReady());
EXPECT_TRUE(f2.hasException()); EXPECT_TRUE(f2.hasException());
} }
TEST(FutureSplitter, splitFuturePriority) {
std::vector<int8_t> priorities = {
folly::Executor::LO_PRI,
folly::Executor::MID_PRI,
folly::Executor::HI_PRI,
};
for (const auto priority : priorities) {
Promise<int> p;
folly::FutureSplitter<int> sp(
p.getSemiFuture().via(&InlineExecutor::instance(), priority));
auto fut = sp.getFuture();
EXPECT_EQ(priority, fut.getPriority());
}
}
...@@ -195,7 +195,9 @@ TEST(Via, chain3) { ...@@ -195,7 +195,9 @@ TEST(Via, chain3) {
} }
struct PriorityExecutor : public Executor { struct PriorityExecutor : public Executor {
void add(Func /* f */) override {} void add(Func /* f */) override {
count1++;
}
void addWithPriority(Func f, int8_t priority) override { void addWithPriority(Func f, int8_t priority) override {
int mid = getNumPriorities() / 2; int mid = getNumPriorities() / 2;
......
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