add pool_executor in pushmi/examples/pool.h, use it when updating the pushmi...

add pool_executor in pushmi/examples/pool.h, use it when updating the pushmi benchmark and migrating it to Folly.Benchmark

Summary: The PushmiBenchmark was badly bit-rotted. Update it for the latest pushmi design, migrate it to Folly.Benchmark, and implement it in terms of a new pushmi::pool_executor that is a thin wrapper over a pointer to a folly::CPUThreadPoolExecutor.

Reviewed By: kirkshoop

Differential Revision: D14550589

fbshipit-source-id: d808d6f5795ead421e6ee88439adb9ee652d3146

folly/experimental/pushmi/benchmarks/PushmiBenchmarks.cpp

@@ -13,6 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+#include <numeric>
+#include <thread>
 #include <vector>
 
 #include <folly/experimental/pushmi/o/defer.h>
@@ -32,7 +35,11 @@
 #include <folly/experimental/pushmi/entangle.h>
 #include <folly/experimental/pushmi/receiver.h>
 
-#include <folly/experimental/pushmi/pool.h>
+#include <folly/experimental/pushmi/examples/pool.h>
+
+#include <folly/Benchmark.h>
+#include <folly/Optional.h>
+#include <folly/container/Foreach.h>
 
 using namespace folly::pushmi::aliases;
 
@@ -47,20 +54,18 @@ struct countdown {
   template <class ExecutorRef>
   void value(ExecutorRef exec);
   template <class E>
-  void error(E e) {
+  void error(E) {
     std::terminate();
   }
   void done() {}
   PUSHMI_TEMPLATE(class Up)
-  (requires mi::Invocable<
-      decltype(mi::set_value),
-      Up,
-      std::ptrdiff_t>)void starting(Up up) {
+  (requires mi::ReceiveValue<Up, std::ptrdiff_t>) //
+  void starting(Up up) {
     mi::set_value(up, 1);
   }
   PUSHMI_TEMPLATE(class Up)
-  (requires mi::True<>&& mi::
-       Invocable<decltype(mi::set_value), Up>)void starting(Up up) volatile {
+  (requires mi::ReceiveValue<Up>&& (!mi::ReceiveValue<Up, std::ptrdiff_t>)) //
+  void starting(Up up) {
     mi::set_value(up);
   }
 };
@@ -69,7 +74,7 @@ template <class R>
 template <class ExecutorRef>
 void countdown<R>::value(ExecutorRef exec) {
   if (--*counter >= 0) {
-    exec | op::submit(R{}(*this));
+    exec.schedule() | op::submit(R{}(*this));
   }
 }
 
@@ -80,37 +85,55 @@ using countdownflowmany = countdown<decltype(mi::make_flow_receiver)>;
 
 struct inline_time_executor {
   using properties = mi::property_set<
-      mi::is_time<>,
       mi::is_executor<>,
-      mi::is_fifo_sequence<>,
+      mi::is_time<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
+    std::chrono::system_clock::time_point at;
+    using properties = mi::property_set<
+        mi::is_sender<>,
         mi::is_always_blocking<>,
         mi::is_single<>>;
 
+    template <class Out>
+    void submit(Out out) {
+      std::this_thread::sleep_until(at);
+      ::mi::set_value(out, inline_time_executor{});
+      ::mi::set_done(out);
+    }
+  };
+
   std::chrono::system_clock::time_point top() {
     return std::chrono::system_clock::now();
   }
-  auto executor() {
-    return *this;
+  auto schedule(std::chrono::system_clock::time_point at) {
+    return task{at};
   }
-  template <class Out>
-  void submit(std::chrono::system_clock::time_point at, Out out) {
-    std::this_thread::sleep_until(at);
-    ::mi::set_value(out, *this);
+  auto schedule() {
+    return schedule(top());
   }
 };
 
 struct inline_executor {
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
     using properties = mi::property_set<
         mi::is_sender<>,
-      mi::is_fifo_sequence<>,
         mi::is_always_blocking<>,
         mi::is_single<>>;
-  auto executor() {
-    return inline_time_executor{};
-  }
+
     template <class Out>
     void submit(Out out) {
-    ::mi::set_value(out, *this);
+      ::mi::set_value(out, inline_executor{});
+      ::mi::set_done(out);
+    }
+  };
+  auto schedule() {
+    return task{};
   }
 };
 
@@ -118,15 +141,19 @@ template <class CancellationFactory>
 struct inline_executor_flow_single {
   CancellationFactory cf;
 
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
+    CancellationFactory cf;
+
     using properties = mi::property_set<
         mi::is_sender<>,
         mi::is_flow<>,
-      mi::is_fifo_sequence<>,
         mi::is_maybe_blocking<>,
         mi::is_single<>>;
-  auto executor() {
-    return inline_time_executor{};
-  }
+
     template <class Out>
     void submit(Out out) {
       auto tokens = cf();
@@ -136,10 +163,10 @@ struct inline_executor_flow_single {
         explicit Data(Stopper stopper) : stopper(std::move(stopper)) {}
         Stopper stopper;
       };
-    auto up = mi::MAKE(receiver)(
+      auto up = mi::make_receiver(
           Data{std::move(tokens.second)},
-        [](auto& data) {},
-        [](auto& data, auto e) noexcept {
+          [](auto&) {},
+          [](auto& data, auto) noexcept {
             auto both = lock_both(data.stopper);
             (*(both.first))(both.second);
           },
@@ -153,12 +180,18 @@ struct inline_executor_flow_single {
 
       auto both = lock_both(tokens.first);
       if (!!both.first && !*(both.first)) {
-      ::mi::set_value(out, *this);
+        ::mi::set_value(out, inline_executor_flow_single{cf});
+        ::mi::set_done(out);
       } else {
         // cancellation is not an error
         ::mi::set_done(out);
       }
     }
+  };
+
+  auto schedule() {
+    return task{cf};
+  }
 };
 
 struct shared_cancellation_factory {
@@ -192,40 +225,50 @@ using inline_executor_flow_single_entangled =
     inline_executor_flow_single<entangled_cancellation_factory>;
 
 struct inline_executor_flow_single_ignore {
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
     using properties = mi::property_set<
         mi::is_sender<>,
         mi::is_flow<>,
-      mi::is_fifo_sequence<>,
         mi::is_maybe_blocking<>,
         mi::is_single<>>;
-  auto executor() {
-    return inline_time_executor{};
-  }
+
     template <class Out>
     void submit(Out out) {
       // pass reference for cancellation.
       ::mi::set_starting(out, mi::receiver<>{});
+      ::mi::set_value(out, inline_executor_flow_single_ignore{});
+      ::mi::set_done(out);
+    }
+  };
 
-    ::mi::set_value(out, *this);
+  auto schedule() {
+    return task{};
   }
 };
 
 struct inline_executor_flow_many {
-  inline_executor_flow_many() : counter(nullptr) {}
+  inline_executor_flow_many() = default;
   inline_executor_flow_many(std::atomic<int>& c) : counter(&c) {}
 
-  std::atomic<int>* counter;
+  std::atomic<int>* counter = nullptr;
+
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
+    std::atomic<int>* counter = nullptr;
 
     using properties = mi::property_set<
         mi::is_sender<>,
         mi::is_flow<>,
-      mi::is_fifo_sequence<>,
         mi::is_maybe_blocking<>,
         mi::is_many<>>;
 
-  auto executor() {
-    return inline_time_executor{};
-  }
     template <class Out>
     void submit(Out out) {
       // boolean cancellation
@@ -241,7 +284,7 @@ struct inline_executor_flow_many {
         std::shared_ptr<producer> p;
       };
 
-    auto up = mi::MAKE(receiver)(
+      auto up = mi::make_receiver(
           Data{p},
           [counter = this->counter](auto& data, auto requested) {
             if (requested < 1) {
@@ -259,7 +302,7 @@ struct inline_executor_flow_many {
               ::mi::set_done(data.p->out);
             }
           },
-        [](auto& data, auto e) noexcept {
+          [](auto& data, auto) noexcept {
             data.p->stop.store(true);
             ::mi::set_done(data.p->out);
           },
@@ -271,324 +314,357 @@ struct inline_executor_flow_many {
       // pass reference for cancellation.
       ::mi::set_starting(p->out, std::move(up));
     }
+  };
+
+  auto schedule() {
+    return task{counter};
+  }
 };
 
 struct inline_executor_flow_many_ignore {
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
     using properties = mi::property_set<
         mi::is_sender<>,
         mi::is_flow<>,
-      mi::is_fifo_sequence<>,
         mi::is_always_blocking<>,
         mi::is_many<>>;
-  auto executor() {
-    return inline_time_executor{};
-  }
     template <class Out>
     void submit(Out out) {
       // pass reference for cancellation.
       ::mi::set_starting(out, mi::receiver<>{});
-
-    ::mi::set_value(out, *this);
-
+      ::mi::set_value(out, inline_executor_flow_many_ignore{});
       ::mi::set_done(out);
     }
+  };
+
+  auto schedule() {
+    return task{};
+  }
 };
 
 struct inline_executor_many {
+  using properties = mi::property_set<
+      mi::is_executor<>,
+      mi::is_fifo_sequence<>>;
+
+  struct task {
     using properties = mi::property_set<
         mi::is_sender<>,
-      mi::is_fifo_sequence<>,
         mi::is_always_blocking<>,
         mi::is_many<>>;
-  auto executor() {
-    return inline_time_executor{};
-  }
     template <class Out>
     void submit(Out out) {
-    ::mi::set_value(out, *this);
+      ::mi::set_value(out, inline_executor_many{});
       ::mi::set_done(out);
     }
-};
+  };
 
-#define concept Concept
-#include <nonius/nonius.h++>
+  auto schedule() {
+    return task{};
+  }
+};
 
-NONIUS_BENCHMARK("ready 1'000 single get (submit)", [](nonius::chronometer meter){
+BENCHMARK(ready_1000_single_get_submit) {
   int counter{0};
-  meter.measure([&]{
   counter = 1'000;
+  std::atomic<int> result{0};
   while (--counter >=0) {
-      auto fortyTwo = op::just(42) | op::get<int>;
+    result += op::just(42) | op::get<int>;
   }
-    return counter;
-  });
-})
+  (void) result.load();
+}
 
-NONIUS_BENCHMARK("ready 1'000 single get (blocking_submit)", [](nonius::chronometer meter){
+BENCHMARK(ready_1000_single_get_blocking_submit) {
   int counter{0};
-  meter.measure([&]{
   counter = 1'000;
+  std::atomic<int> result{0};
   while (--counter >=0) {
-      auto fortyTwo = mi::make_single_sender([](auto out){ mi::set_value(out, 42); mi::set_done(out);}) | op::get<int>;
+    result += mi::make_single_sender(
+      [](auto out){
+        mi::set_value(out, 42);
+        mi::set_done(out);
+      })
+    | op::get<int>;
   }
-    return counter;
-  });
-})
+  (void) result.load();
+}
 
-NONIUS_BENCHMARK("inline 1'000 single", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_single, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor{};
   using IE = decltype(ie);
   countdownsingle single{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_receiver(single));
+    ie.schedule() | op::submit(mi::make_receiver(single));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 time single", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_time_single, n) {
   std::atomic<int> counter{0};
   auto ie = inline_time_executor{};
   using IE = decltype(ie);
   countdownsingle single{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_receiver(single));
+    ie.schedule() | op::submit(mi::make_receiver(single));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 many", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_many, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_many{};
   using IE = decltype(ie);
   countdownmany many{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_receiver(many));
+    ie.schedule() | op::submit(mi::make_receiver(many));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 flow_single shared", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_flow_single_shared, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_single_shared{};
   using IE = decltype(ie);
   countdownflowsingle flowsingle{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(flowsingle));
+    ie.schedule() | op::submit(mi::make_flow_receiver(flowsingle));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 flow_single entangle", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_flow_single_entangle, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_single_entangled{};
   using IE = decltype(ie);
   countdownflowsingle flowsingle{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(flowsingle));
+    ie.schedule() | op::submit(mi::make_flow_receiver(flowsingle));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 flow_single ignore cancellation", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_flow_single_ignore_cancellation, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_single_ignore{};
   using IE = decltype(ie);
   countdownflowsingle flowsingle{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(flowsingle));
+    ie.schedule() | op::submit(mi::make_flow_receiver(flowsingle));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 flow_many", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_flow_many, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_many{};
   using IE = decltype(ie);
   countdownflowmany flowmany{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(flowmany));
+    ie.schedule() | op::submit(mi::make_flow_receiver(flowmany));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1 flow_many with 1'000 values pull 1", [](nonius::chronometer meter){
+BENCHMARK(inline_1_flow_many_with_1000_values_pull_1, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_many{counter};
   using IE = decltype(ie);
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::for_each(mi::make_receiver());
+    ie.schedule() | op::for_each(mi::make_receiver());
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1 flow_many with 1'000 values pull 1'000", [](nonius::chronometer meter){
+BENCHMARK(inline_1_flow_many_with_1000_values_pull_1000, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_many{counter};
   using IE = decltype(ie);
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(mi::ignoreNF{}, mi::abortEF{}, mi::ignoreDF{}, [](auto up){
+    ie.schedule()
+      | op::submit(
+          mi::make_flow_receiver(
+            mi::ignoreNF{},
+            mi::abortEF{},
+            mi::ignoreDF{},
+            [](auto up){
               mi::set_value(up, 1'000);
-    }));
+            }
+          )
+        );
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("inline 1'000 flow_many ignore cancellation", [](nonius::chronometer meter){
+BENCHMARK(inline_1000_flow_many_ignore_cancellation, n) {
   std::atomic<int> counter{0};
   auto ie = inline_executor_flow_many_ignore{};
   using IE = decltype(ie);
   countdownflowmany flowmany{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    ie | op::submit(mi::make_flow_receiver(flowmany));
+    ie.schedule() | op::submit(mi::make_flow_receiver(flowmany));
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("trampoline 1'000 single get (blocking_submit)", [](nonius::chronometer meter){
+BENCHMARK(trampoline_1000_single_get_blocking_submit, n) {
   int counter{0};
   auto tr = mi::trampoline();
   using TR = decltype(tr);
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter = 1'000;
-    while (--counter >=0) {
-      auto fortyTwo = tr | op::transform([](auto){return 42;}) | op::get<int>;
+    while (--counter >= 0) {
+      auto fortyTwo = tr.schedule()
+        | op::transform([](auto){return 42;})
+        | op::get<int>;
+      (void) fortyTwo;
     }
-    return counter;
-  });
-})
+    (void) counter;
+  }
+}
 
-NONIUS_BENCHMARK("trampoline static derecursion 1'000", [](nonius::chronometer meter){
+BENCHMARK(trampoline_static_derecursion_1000, n) {
   std::atomic<int> counter{0};
   auto tr = mi::trampoline();
   using TR = decltype(tr);
   countdownsingle single{counter};
-  meter.measure([&]{
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    tr | op::submit(single);
+    tr.schedule() | op::submit(single);
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("trampoline virtual derecursion 1'000", [](nonius::chronometer meter){
+BENCHMARK(trampoline_virtual_derecursion_1000, n) {
   std::atomic<int> counter{0};
   auto tr = mi::trampoline();
   using TR = decltype(tr);
   auto single = countdownsingle{counter};
   std::function<void(mi::any_executor_ref<>)> recurse{
-      [&](auto exec) { ::folly::pushmi::set_value(single, exec); }};
-  meter.measure([&]{
+      [&](auto exec) { mi::set_value(single, exec); }};
+
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    tr | op::submit([&](auto exec) { recurse(exec); });
+    tr.schedule() | op::submit([&](auto exec) { recurse(exec); });
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("trampoline flow_many_sender 1'000", [](nonius::chronometer meter){
+BENCHMARK(trampoline_flow_many_sender_1000, n) {
   std::atomic<int> counter{0};
   auto tr = mi::trampoline();
   using TR = decltype(tr);
   std::vector<int> values(1'000);
+  BENCHMARK_SUSPEND {
     std::iota(values.begin(), values.end(), 1);
+  }
   auto f = op::flow_from(values, tr) | op::tap([&](int){
     --counter;
   });
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
     f | op::for_each(mi::make_receiver());
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("pool{1} submit 1'000", [](nonius::chronometer meter){
-  mi::pool pl{std::max(1u,std::thread::hardware_concurrency())};
-  auto pe = pl.executor();
-  using PE = decltype(pe);
+BENCHMARK(pool_1_submit_1000, n) {
+  folly::Optional<mi::pool> pl;
+  mi::pool_executor pe;
+  BENCHMARK_SUSPEND {
+    pl.emplace(1u);
+    pe = pl->executor();
+  }
   std::atomic<int> counter{0};
   countdownsingle single{counter};
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    pe | op::submit(single);
+    pe.schedule() | op::submit(single);
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("pool{hardware_concurrency} submit 1'000", [](nonius::chronometer meter){
-  mi::pool pl{std::min(1u,std::thread::hardware_concurrency())};
-  auto pe = pl.executor();
-  using PE = decltype(pe);
+BENCHMARK(pool_hardware_concurrency_submit_1000, n) {
+  folly::Optional<mi::pool> pl;
+  mi::pool_executor pe;
+  BENCHMARK_SUSPEND {
+    pl.emplace(std::max(1u, std::thread::hardware_concurrency()));
+    pe = pl->executor();
+  }
   std::atomic<int> counter{0};
   countdownsingle single{counter};
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    pe | op::submit(single);
+    pe.schedule() | op::submit(single);
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("new thread submit 1'000", [](nonius::chronometer meter){
+BENCHMARK(new_thread_submit_1000, n) {
   auto nt = mi::new_thread();
   using NT = decltype(nt);
   std::atomic<int> counter{0};
   countdownsingle single{counter};
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    nt | op::submit(single);
+    nt.schedule() | op::submit(single);
     while(counter.load() > 0);
-    return counter.load();
-  });
-})
+  }
+}
 
-NONIUS_BENCHMARK("new thread blocking_submit 1'000", [](nonius::chronometer meter){
+BENCHMARK(new_thread_blocking_submit_1000, n) {
   auto nt = mi::new_thread();
-  using NT = decltype(nt);
   std::atomic<int> counter{0};
   countdownsingle single{counter};
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    nt | op::blocking_submit(single);
-    return counter.load();
-  });
-})
+    nt.schedule() | op::blocking_submit(single);
+  }
+}
 
-NONIUS_BENCHMARK("new thread + time submit 1'000", [](nonius::chronometer meter){
+BENCHMARK(new_thread_and_time_submit_1000, n) {
   auto nt = mi::new_thread();
-  using NT = decltype(nt);
   auto time = mi::time_source<>{};
-  auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })();
-  using TNT = decltype(tnt);
+  auto strands = time.make(mi::systemNowF{}, nt);
+  auto tnt = mi::make_strand(strands);
   std::atomic<int> counter{0};
   countdownsingle single{counter};
-  meter.measure([&]{
+  FOR_EACH_RANGE (i, 0, n) {
     counter.store(1'000);
-    tnt | op::submit(single);
+    tnt.schedule() | op::submit(single);
     while(counter.load() > 0);
-    return counter.load();
-  });
+  }
+  BENCHMARK_SUSPEND {
     time.join();
-})
+  }
+}
+
+int main() {
+  folly::runBenchmarks();
+}

folly/experimental/pushmi/examples/pool.h

+/*
+ * Copyright 2018-present 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/experimental/pushmi/properties.h>
+#include <folly/experimental/pushmi/concepts.h>
+
+#include <folly/Executor.h>
+#include <folly/executors/CPUThreadPoolExecutor.h>
+
+namespace folly {
+namespace pushmi {
+
+class pool;
+
+class pool_executor {
+  struct task;
+  Executor::KeepAlive<CPUThreadPoolExecutor> exec_ {};
+
+public:
+  using properties =
+    property_set<is_executor<>, is_concurrent_sequence<>>;
+
+  pool_executor() = default;
+  explicit pool_executor(pool &e);
+  task schedule();
+};
+
+struct pool_executor::task {
+  using properties =
+    property_set<
+      is_sender<>, is_never_blocking<>, is_single<>>;
+
+  explicit task(pool_executor e)
+    : pool_ex_(std::move(e))
+  {}
+
+  PUSHMI_TEMPLATE(class Out)
+    (requires ReceiveValue<Out, pool_executor&>)
+  void submit(Out out) && {
+    pool_ex_.exec_->add([e = pool_ex_, out = std::move(out)]() mutable {
+      set_value(out, e);
+      set_done(out);
+    });
+  }
+private:
+  pool_executor pool_ex_;
+};
+
+class pool {
+  friend pool_executor;
+  CPUThreadPoolExecutor pool_;
+
+public:
+  explicit pool(std::size_t threads) : pool_(threads) {}
+
+  auto executor() {
+    return pool_executor{*this};
+  }
+
+  void stop() {
+    pool_.stop();
+  }
+
+  void wait() {
+    pool_.join();
+  }
+};
+
+inline pool_executor::pool_executor(pool &e)
+: exec_(Executor::getKeepAliveToken(e.pool_))
+{}
+
+inline pool_executor::task pool_executor::schedule() {
+  return task{*this};
+}
+
+} // namespace pushmi
+} // namespace folly