RFC: future executor

Summary:
not necessarily for commit but i was playing around a bit with this today to see what it might look like, so i figured i'd put it up
i assume this shenanigan isn't remotely safe (threadwise) without some extra magic... @fugalh we chatted about this last time i was in mpk a bit.
maybe addFuture could take an executor to make sure the promise is fulfilled on the correct thread, or something.

you'd have to reimplement timeouts via this executor or via futures if you wanted to get them via the futures channel, which makes sense anyway as this could be used with any Executor and not just ThreadPoolExecutor which owns that implementation

also specialized makeFutureTry to take functions that return futures but treat them like they return the contained types, so fulfil() can be used liked then()
this specialization could just as easily be done on fulfil() itself if we wanted.

Test Plan: unit

Reviewed By: davejwatson@fb.com

Subscribers: hannesr, trunkagent, craffert, njormrod, fugalh, bmatheny

FB internal diff: D1566369

diff --git a/folly/experimental/wangle/concurrent/FutureExecutor.h b/folly/experimental/wangle/concurrent/FutureExecutor.h
new file mode 100644 (file)
index 0000000..ecf392b
--- /dev/null
+++ b/folly/experimental/wangle/concurrent/FutureExecutor.h
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2014 Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <folly/wangle/Future.h>
+
+namespace folly { namespace wangle {
+
+template <typename ExecutorImpl>
+class FutureExecutor : public ExecutorImpl {
+ public:
+  template <typename... Args>
+  explicit FutureExecutor(Args&&... args)
+    : ExecutorImpl(std::forward<Args>(args)...) {}
+
+  /*
+   * Given a function func that returns a Future<T>, adds that function to the
+   * contained Executor and returns a Future<T> which will be fulfilled with
+   * func's result once it has been executed.
+   *
+   * For example: auto f = futureExecutor.addFuture([](){
+   *                return doAsyncWorkAndReturnAFuture();
+   *              });
+   */
+  template <typename F>
+  typename std::enable_if<isFuture<typename std::result_of<F()>::type>::value,
+                          typename std::result_of<F()>::type>::type
+  addFuture(F func) {
+    typedef typename std::result_of<F()>::type::value_type T;
+    Promise<T> promise;
+    auto future = promise.getFuture();
+    auto movePromise = folly::makeMoveWrapper(std::move(promise));
+    auto moveFunc = folly::makeMoveWrapper(std::move(func));
+    ExecutorImpl::add([movePromise, moveFunc] () mutable {
+      (*moveFunc)().then([movePromise] (Try<T>&& t) mutable {
+        movePromise->fulfilTry(std::move(t));
+      });
+    });
+    return future;
+  }
+
+  /*
+   * Similar to addFuture above, but takes a func that returns some non-Future
+   * type T.
+   *
+   * For example: auto f = futureExecutor.addFuture([]() {
+   *                return 42;
+   *              });
+   */
+  template <typename F>
+  typename std::enable_if<!isFuture<typename std::result_of<F()>::type>::value,
+                          Future<typename std::result_of<F()>::type>>::type
+  addFuture(F func) {
+    typedef typename std::result_of<F()>::type T;
+    Promise<T> promise;
+    auto future = promise.getFuture();
+    auto movePromise = folly::makeMoveWrapper(std::move(promise));
+    auto moveFunc = folly::makeMoveWrapper(std::move(func));
+    ExecutorImpl::add([movePromise, moveFunc] () mutable {
+      movePromise->fulfil(std::move(*moveFunc));
+    });
+    return future;
+  }
+};
+
+}}

diff --git a/folly/experimental/wangle/concurrent/test/ThreadPoolExecutorTest.cpp b/folly/experimental/wangle/concurrent/test/ThreadPoolExecutorTest.cpp
index bacee2ba4a87867021e157132148d28fb061b18e..00d5ccac2588106f28efee2d5bca71cf74bb3282 100644 (file)
--- a/folly/experimental/wangle/concurrent/test/ThreadPoolExecutorTest.cpp
+++ b/folly/experimental/wangle/concurrent/test/ThreadPoolExecutorTest.cpp
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include <folly/experimental/wangle/concurrent/FutureExecutor.h>
 #include <folly/experimental/wangle/concurrent/ThreadPoolExecutor.h>
 #include <folly/experimental/wangle/concurrent/CPUThreadPoolExecutor.h>
 #include <folly/experimental/wangle/concurrent/IOThreadPoolExecutor.h>
@@ -217,3 +218,58 @@ TEST(ThreadPoolExecutorTest, CPUExpiration) {
 TEST(ThreadPoolExecutorTest, IOExpiration) {
   expiration<IOThreadPoolExecutor>();
 }
+
+template <typename TPE>
+static void futureExecutor() {
+  FutureExecutor<TPE> fe(2);
+  int c = 0;
+  fe.addFuture([] () { return makeFuture<int>(42); }).then(
+    [&] (Try<int>&& t) {
+      c++;
+      EXPECT_EQ(42, t.value());
+    });
+  fe.addFuture([] () { return 100; }).then(
+    [&] (Try<int>&& t) {
+      c++;
+      EXPECT_EQ(100, t.value());
+    });
+  fe.addFuture([] () { return makeFuture(); }).then(
+    [&] (Try<void>&& t) {
+      c++;
+      EXPECT_NO_THROW(t.value());
+    });
+  fe.addFuture([] () { return; }).then(
+    [&] (Try<void>&& t) {
+      c++;
+      EXPECT_NO_THROW(t.value());
+    });
+  fe.addFuture([] () { throw std::runtime_error("oops"); }).then(
+    [&] (Try<void>&& t) {
+      c++;
+      EXPECT_THROW(t.value(), std::runtime_error);
+    });
+  // Test doing actual async work
+  fe.addFuture([] () {
+    auto p = std::make_shared<Promise<int>>();
+    std::thread t([p](){
+      burnMs(10)();
+      p->setValue(42);
+    });
+    t.detach();
+    return p->getFuture();
+  }).then([&] (Try<int>&& t) {
+    EXPECT_EQ(42, t.value());
+    c++;
+  });
+  burnMs(15)(); // Sleep long enough for the promise to be fulfilled
+  fe.join();
+  EXPECT_EQ(6, c);
+}
+
+TEST(ThreadPoolExecutorTest, CPUFuturePool) {
+  futureExecutor<CPUThreadPoolExecutor>();
+}
+
+TEST(ThreadPoolExecutorTest, IOFuturePool) {
+  futureExecutor<IOThreadPoolExecutor>();
+}