Commit 5a67fac7 authored by Lee Howes's avatar Lee Howes Committed by Facebook GitHub Bot

Replace default executor argument in FutureDAG

Summary:
Change to SemiFuture returning collect, which checks for nullptr passed to via, breaks this test which produced a default nullptr.

Instead remove default and force passing of a valid executor.

Reviewed By: yfeldblum

Differential Revision: D20795552

fbshipit-source-id: 5411fea910d9f85eb5a4ba825a1cfd48a72b8a44
parent c1beec58
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#pragma once #pragma once
#include <folly/Executor.h>
#include <folly/futures/Future.h> #include <folly/futures/Future.h>
#include <folly/futures/SharedPromise.h> #include <folly/futures/SharedPromise.h>
...@@ -23,14 +24,16 @@ namespace folly { ...@@ -23,14 +24,16 @@ namespace folly {
class FutureDAG : public std::enable_shared_from_this<FutureDAG> { class FutureDAG : public std::enable_shared_from_this<FutureDAG> {
public: public:
static std::shared_ptr<FutureDAG> create() { static std::shared_ptr<FutureDAG> create(
return std::shared_ptr<FutureDAG>(new FutureDAG()); Executor::KeepAlive<> defaultExecutor) {
return std::shared_ptr<FutureDAG>(
new FutureDAG(std::move(defaultExecutor)));
} }
typedef size_t Handle; typedef size_t Handle;
typedef std::function<Future<Unit>()> FutureFunc; typedef std::function<Future<Unit>()> FutureFunc;
Handle add(FutureFunc func, Executor* executor = nullptr) { Handle add(FutureFunc func, Executor::KeepAlive<> executor) {
nodes.emplace_back(std::move(func), executor); nodes.emplace_back(std::move(func), executor);
return nodes.size() - 1; return nodes.size() - 1;
} }
...@@ -106,12 +109,12 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> { ...@@ -106,12 +109,12 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> {
} }
} }
auto sinkHandle = add([] { return Future<Unit>(); }); auto sinkHandle = add([] { return Future<Unit>(); }, defaultExecutor_);
for (auto handle : leafNodes) { for (auto handle : leafNodes) {
dependency(handle, sinkHandle); dependency(handle, sinkHandle);
} }
auto sourceHandle = add(nullptr); auto sourceHandle = add(nullptr, defaultExecutor_);
for (auto handle : rootNodes) { for (auto handle : rootNodes) {
dependency(sourceHandle, handle); dependency(sourceHandle, handle);
} }
...@@ -142,7 +145,8 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> { ...@@ -142,7 +145,8 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> {
} }
private: private:
FutureDAG() = default; FutureDAG(Executor::KeepAlive<> defaultExecutor)
: defaultExecutor_{std::move(defaultExecutor)} {}
bool hasCycle() { bool hasCycle() {
// Perform a modified topological sort to detect cycles // Perform a modified topological sort to detect cycles
...@@ -187,11 +191,11 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> { ...@@ -187,11 +191,11 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> {
} }
struct Node { struct Node {
Node(FutureFunc&& funcArg, Executor* executorArg) Node(FutureFunc&& funcArg, Executor::KeepAlive<> executorArg)
: func(std::move(funcArg)), executor(executorArg) {} : func(std::move(funcArg)), executor(std::move(executorArg)) {}
FutureFunc func{nullptr}; FutureFunc func{nullptr};
Executor* executor{nullptr}; Executor::KeepAlive<> executor;
SharedPromise<Unit> promise; SharedPromise<Unit> promise;
std::vector<Handle> dependencies; std::vector<Handle> dependencies;
bool hasDependents{false}; bool hasDependents{false};
...@@ -199,13 +203,14 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> { ...@@ -199,13 +203,14 @@ class FutureDAG : public std::enable_shared_from_this<FutureDAG> {
}; };
std::vector<Node> nodes; std::vector<Node> nodes;
Executor::KeepAlive<> defaultExecutor_;
}; };
// Polymorphic functor implementation // Polymorphic functor implementation
template <typename T> template <typename T>
class FutureDAGFunctor { class FutureDAGFunctor {
public: public:
std::shared_ptr<FutureDAG> dag = FutureDAG::create(); std::shared_ptr<FutureDAG> dag;
T state; T state;
std::vector<T> dep_states; std::vector<T> dep_states;
T result() { T result() {
...@@ -220,8 +225,10 @@ class FutureDAGFunctor { ...@@ -220,8 +225,10 @@ class FutureDAGFunctor {
this->dag->go().get(); this->dag->go().get();
} }
virtual void operator()() {} virtual void operator()() {}
explicit FutureDAGFunctor(T init_val) : state(init_val) {} explicit FutureDAGFunctor(T init_val, Executor::KeepAlive<> defaultExecutor)
FutureDAGFunctor() : state() {} : dag(FutureDAG::create(std::move(defaultExecutor))), state(init_val) {}
FutureDAGFunctor(Executor::KeepAlive<> defaultExecutor)
: dag(FutureDAG::create(std::move(defaultExecutor))), state() {}
virtual ~FutureDAGFunctor() {} virtual ~FutureDAGFunctor() {}
}; };
......
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#include <folly/experimental/FutureDAG.h> #include <folly/experimental/FutureDAG.h>
#include <boost/thread/barrier.hpp> #include <boost/thread/barrier.hpp>
#include <folly/executors/GlobalExecutor.h>
#include <folly/portability/GTest.h> #include <folly/portability/GTest.h>
using namespace folly; using namespace folly;
...@@ -87,14 +88,15 @@ struct FutureDAGTest : public testing::Test { ...@@ -87,14 +88,15 @@ struct FutureDAGTest : public testing::Test {
test->order.push_back(handle); test->order.push_back(handle);
return Future<Unit>(); return Future<Unit>();
}), }),
handle(test->dag->add(func)) {} handle(test->dag->add(func, getGlobalCPUExecutor())) {}
const FutureDAG::FutureFunc func; const FutureDAG::FutureFunc func;
const Handle handle; const Handle handle;
std::set<Handle> dependencies; std::set<Handle> dependencies;
}; };
const std::shared_ptr<FutureDAG> dag = FutureDAG::create(); const std::shared_ptr<FutureDAG> dag =
FutureDAG::create(getGlobalCPUExecutor());
std::map<Handle, std::unique_ptr<TestNode>> nodes; std::map<Handle, std::unique_ptr<TestNode>> nodes;
std::vector<Handle> order; std::vector<Handle> order;
}; };
...@@ -216,15 +218,15 @@ FutureDAG::FutureFunc throwFunc = [] { ...@@ -216,15 +218,15 @@ FutureDAG::FutureFunc throwFunc = [] {
}; };
TEST_F(FutureDAGTest, ThrowBegin) { TEST_F(FutureDAGTest, ThrowBegin) {
auto h1 = dag->add(throwFunc); auto h1 = dag->add(throwFunc, getGlobalCPUExecutor());
auto h2 = dag->add(makeFutureFunc); auto h2 = dag->add(makeFutureFunc, getGlobalCPUExecutor());
dag->dependency(h1, h2); dag->dependency(h1, h2);
EXPECT_THROW(dag->go().get(), std::runtime_error); EXPECT_THROW(dag->go().get(), std::runtime_error);
} }
TEST_F(FutureDAGTest, ThrowEnd) { TEST_F(FutureDAGTest, ThrowEnd) {
auto h1 = dag->add(makeFutureFunc); auto h1 = dag->add(makeFutureFunc, getGlobalCPUExecutor());
auto h2 = dag->add(throwFunc); auto h2 = dag->add(throwFunc, getGlobalCPUExecutor());
dag->dependency(h1, h2); dag->dependency(h1, h2);
EXPECT_THROW(dag->go().get(), std::runtime_error); EXPECT_THROW(dag->go().get(), std::runtime_error);
} }
...@@ -257,17 +259,19 @@ TEST_F(FutureDAGTest, DestroyBeforeComplete) { ...@@ -257,17 +259,19 @@ TEST_F(FutureDAGTest, DestroyBeforeComplete) {
auto barrier = std::make_shared<boost::barrier>(2); auto barrier = std::make_shared<boost::barrier>(2);
Future<Unit> f; Future<Unit> f;
{ {
auto localDag = FutureDAG::create(); auto localDag = FutureDAG::create(getGlobalCPUExecutor());
auto h1 = localDag->add([barrier] { auto h1 = localDag->add(
auto p = std::make_shared<Promise<Unit>>(); [barrier] {
std::thread t([p, barrier] { auto p = std::make_shared<Promise<Unit>>();
barrier->wait(); std::thread t([p, barrier] {
p->setValue(); barrier->wait();
}); p->setValue();
t.detach(); });
return p->getFuture(); t.detach();
}); return p->getFuture();
auto h2 = localDag->add(makeFutureFunc); },
getGlobalCPUExecutor());
auto h2 = localDag->add(makeFutureFunc, getGlobalCPUExecutor());
localDag->dependency(h1, h2); localDag->dependency(h1, h2);
f = localDag->go(); f = localDag->go();
} }
......
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