Commit b6f62e95 authored by Hans Fugal's avatar Hans Fugal Committed by Andrii Grynenko

Fix via

Summary: Sometimes you just have to take a step back. :-P

Test Plan: All the unit tests including the one that had been disabled, now pass.

Reviewed By: hannesr@fb.com

Subscribers: meisner, trunkagent, net-systems@, fugalh, exa, njormrod, davejwatson

FB internal diff: D1596368

Tasks: 4920689, 4480567, 5306911
parent 853d5b59
...@@ -193,42 +193,11 @@ template <class T> ...@@ -193,42 +193,11 @@ template <class T>
template <typename Executor> template <typename Executor>
inline Future<T> Future<T>::via(Executor* executor) { inline Future<T> Future<T>::via(Executor* executor) {
throwIfInvalid(); throwIfInvalid();
MoveWrapper<Promise<T>> promise;
auto f = promise->getFuture();
// We are obligated to return a cold future.
f.deactivate();
// But we also need to make this one cold for via to at least work some of
// the time. (see below)
deactivate();
then([=](Try<T>&& t) mutable {
MoveWrapper<Try<T>> tw(std::move(t));
// There is a race here.
// When the promise is fulfilled, and the future is still inactive, when
// the future is activated (e.g. by destruction) the callback will happen
// in that context, not in the intended context (which has already left
// the building).
//
// Currently, this will work fine because all the temporaries are
// destructed in an order that is compatible with this implementation:
//
// makeFuture().via(x).then(a).then(b);
//
// However, this will not work reliably:
//
// auto f2 = makeFuture().via(x);
// f2.then(a).then(b);
//
// Because the first temporary is destructed on the first line, and the
// executor is fed. But by the time f2 is destructed, the executor
// may have already fulfilled the promise on the other thread.
//
// TODO(#4920689) fix it
executor->add([=]() mutable { promise->fulfilTry(std::move(*tw)); });
});
return f; this->deactivate();
state_->setExecutor(executor);
return std::move(*this);
} }
template <class T> template <class T>
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include <folly/wangle/Try.h> #include <folly/wangle/Try.h>
#include <folly/wangle/Promise.h> #include <folly/wangle/Promise.h>
#include <folly/wangle/Future.h> #include <folly/wangle/Future.h>
#include <folly/wangle/Executor.h>
namespace folly { namespace wangle { namespace detail { namespace folly { namespace wangle { namespace detail {
...@@ -141,14 +142,26 @@ class State { ...@@ -141,14 +142,26 @@ class State {
bool isActive() { return active_; } bool isActive() { return active_; }
void setExecutor(Executor* x) {
std::lock_guard<decltype(mutex_)> lock(mutex_);
executor_ = x;
}
private: private:
void maybeCallback() { void maybeCallback() {
std::lock_guard<decltype(mutex_)> lock(mutex_); std::lock_guard<decltype(mutex_)> lock(mutex_);
if (!calledBack_ && if (!calledBack_ &&
value_ && callback_ && isActive()) { value_ && callback_ && isActive()) {
// TODO we should probably try/catch here // TODO(5306911) we should probably try/catch here
callback_(std::move(*value_)); if (executor_) {
calledBack_ = true; MoveWrapper<folly::Optional<Try<T>>> val(std::move(value_));
MoveWrapper<std::function<void(Try<T>&&)>> cb(std::move(callback_));
executor_->add([cb, val]() mutable { (*cb)(std::move(**val)); });
calledBack_ = true;
} else {
callback_(std::move(*value_));
calledBack_ = true;
}
} }
} }
...@@ -173,6 +186,7 @@ class State { ...@@ -173,6 +186,7 @@ class State {
bool calledBack_ = false; bool calledBack_ = false;
unsigned char detached_ = 0; unsigned char detached_ = 0;
bool active_ = true; bool active_ = true;
Executor* executor_ = nullptr;
// this lock isn't meant to protect all accesses to members, only the ones // this lock isn't meant to protect all accesses to members, only the ones
// that need to be threadsafe: the act of setting value_ and callback_, and // that need to be threadsafe: the act of setting value_ and callback_, and
......
...@@ -808,7 +808,7 @@ TEST(Future, viaRaces) { ...@@ -808,7 +808,7 @@ TEST(Future, viaRaces) {
} }
// TODO(#4920689) // TODO(#4920689)
TEST(Future, DISABLED_viaRaces_2stage) { TEST(Future, viaRaces_2stage) {
ManualExecutor x; ManualExecutor x;
Promise<void> p; Promise<void> p;
auto tid = std::this_thread::get_id(); auto tid = std::this_thread::get_id();
......
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