Async fork/join

CMakeLists.txt

@@ -73,7 +73,6 @@ endif ()
 add_custom_target(buildSingleHeader
     COMMAND ${CMAKE_COMMAND} -P ${CMAKE_CURRENT_SOURCE_DIR}/buildSingleHeader.cmake
     WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
-add_dependencies(pushmi buildSingleHeader)
 
 install(
     DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/

Readme.md

@@ -157,121 +157,121 @@ auto s0 = single<int>{single{}};
 auto s1 = single<int, std::exception_ptr>{single{}};
 ```
 
-## `deferred`
+## `sender`
 
-The `deferred` type in the library provides simple ways to construct new implementations of the NoneSender concept.
+The `sender` type in the library provides simple ways to construct new implementations of the NoneSender concept.
 
 construct a producer of nothing, aka `never()`
 
 ```cpp
-deferred<> d;
+sender<> d;
 ```
 
 construct new type using one or more lambdas, or with designated initializers, use multiple lambdas to build overload sets
 
 ```cpp
-auto d0 = deferred{on_submit{[](auto out){}}};
-auto d1 = deferred{[](auto out){}};
-auto d2 = deferred{on_submit{[](none<> out){}, [](auto out){}}};
+auto d0 = sender{on_submit{[](auto out){}}};
+auto d1 = sender{[](auto out){}};
+auto d2 = sender{on_submit{[](none<> out){}, [](auto out){}}};
 
 ```
 
-construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing deferred. The state must be a NoneSender, but can be a super-set with additional state for this filter.
+construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing sender. The state must be a NoneSender, but can be a super-set with additional state for this filter.
 
 ```cpp
-auto d0 = deferred{deferred{}};
+auto d0 = sender{sender{}};
 
-auto d1 = deferred{deferred{}, on_submit{
-    [](deferred<>& in, auto out){in | submit(out);}}};
+auto d1 = sender{sender{}, on_submit{
+    [](sender<>& in, auto out){in | submit(out);}}};
 
-auto d2 = deferred{deferred{},
-    [](deferred<>& in, auto out){in | submit(out);}};
+auto d2 = sender{sender{},
+    [](sender<>& in, auto out){in | submit(out);}};
 
 ```
 
 construct a type-erased type for a particular E (which could be a std::variant of supported types). I have a plan to provide operators to collapse values and errors to variant or tuple and then expand from variant or tuple back to their constituent values/errors.
 
 ```cpp
-auto d0 = deferred<>{deferred{}};
-auto d1 = deferred<std::exception_ptr>{deferred{}};
+auto d0 = sender<>{sender{}};
+auto d1 = sender<std::exception_ptr>{sender{}};
 ```
 
-## `single_deferred`
+## `single_sender`
 
-The `single_deferred` type in the library provides simple ways to construct new implementations of the SingleSender concept.
+The `single_sender` type in the library provides simple ways to construct new implementations of the SingleSender concept.
 
 construct a producer of nothing, aka `never()`
 
 ```cpp
-single_deferred<> sd;
+single_sender<> sd;
 ```
 
 construct new type using one or more lambdas, or with designated initializers, use multiple lambdas to build overload sets
 
 ```cpp
-auto sd0 = single_deferred{on_submit{[](auto out){}}};
-auto sd1 = single_deferred{[](auto out){}};
-auto sd2 = single_deferred{on_submit{[](single<> out){}, [](auto out){}}};
+auto sd0 = single_sender{on_submit{[](auto out){}}};
+auto sd1 = single_sender{[](auto out){}};
+auto sd2 = single_sender{on_submit{[](single<> out){}, [](auto out){}}};
 
 ```
 
-construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing single_deferred. The state must be a SingleSender, but can be a super-set with additional state for this filter.
+construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing single_sender. The state must be a SingleSender, but can be a super-set with additional state for this filter.
 
 ```cpp
-auto sd0 = single_deferred{single_deferred{}};
+auto sd0 = single_sender{single_sender{}};
 
-auto sd1 = single_deferred{single_deferred{}, on_submit{
-    [](single_deferred<>& in, auto out){in | submit(out);}}};
+auto sd1 = single_sender{single_sender{}, on_submit{
+    [](single_sender<>& in, auto out){in | submit(out);}}};
 
-auto sd2 = single_deferred{single_deferred{},
-    [](single_deferred<>& in, auto out){in | submit(out);}};
+auto sd2 = single_sender{single_sender{},
+    [](single_sender<>& in, auto out){in | submit(out);}};
 
 ```
 
 construct a type-erased type for a particular T & E (which could be a std::variant of supported types). I have a plan to provide operators to collapse values and errors to variant or tuple and then expand from variant or tuple back to their constituent values/errors.
 
 ```cpp
-auto sd0 = single_deferred<int>{single_deferred{}};
-auto sd1 = single_deferred<int, std::exception_ptr>{single_deferred{}};
+auto sd0 = single_sender<int>{single_sender{}};
+auto sd1 = single_sender<int, std::exception_ptr>{single_sender{}};
 ```
 
-## `time_single_deferred`
+## `time_single_sender`
 
-The `time_single_deferred` type in the library provides simple ways to construct new implementations of the TimeSingleSender concept.
+The `time_single_sender` type in the library provides simple ways to construct new implementations of the TimeSingleSender concept.
 
 construct a producer of nothing, aka `never()`
 
 ```cpp
-time_single_deferred<> tsd;
+time_single_sender<> tsd;
 ```
 
 construct new type using one or more lambdas, or with designated initializers, use multiple lambdas to build overload sets
 
 ```cpp
-auto tsd0 = time_single_deferred{on_submit{[](auto at, auto out){}}};
-auto tsd1 = time_single_deferred{[](auto at, auto out){}};
-auto tsd2 = time_single_deferred{on_submit{[](auto at, single<> out){}, [](auto at, auto out){}}};
+auto tsd0 = time_single_sender{on_submit{[](auto at, auto out){}}};
+auto tsd1 = time_single_sender{[](auto at, auto out){}};
+auto tsd2 = time_single_sender{on_submit{[](auto at, single<> out){}, [](auto at, auto out){}}};
 
 ```
 
-construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing time_single_deferred. The state must be a SingleSender, but can be a super-set with additional state for this filter.
+construct a new type with shared state across the lambdas. very useful for building a filter on top of an existing time_single_sender. The state must be a SingleSender, but can be a super-set with additional state for this filter.
 
 ```cpp
-auto tsd0 = time_single_deferred{single_deferred{}};
+auto tsd0 = time_single_sender{single_sender{}};
 
-auto tsd1 = time_single_deferred{single_deferred{}, on_submit{
-    [](time_single_deferred<>& in, auto at, auto out){in | submit(at, out);}}};
+auto tsd1 = time_single_sender{single_sender{}, on_submit{
+    [](time_single_sender<>& in, auto at, auto out){in | submit(at, out);}}};
 
-auto tsd2 = time_single_deferred{single_deferred{},
-    [](time_single_deferred<>& in, auto at, auto out){in | submit(at, out);}};
+auto tsd2 = time_single_sender{single_sender{},
+    [](time_single_sender<>& in, auto at, auto out){in | submit(at, out);}};
 
 ```
 
 construct a type-erased type for a particular T & E (which could be a std::variant of supported types). I have a plan to provide operators to collapse values and errors to variant or tuple and then expand from variant or tuple back to their constituent values/errors.
 
 ```cpp
-auto tsd0 = time_single_deferred<int>{time_single_deferred{}};
-auto tsd1 = time_single_deferred<int, std::exception_ptr>{time_single_deferred{}};
+auto tsd0 = time_single_sender<int>{time_single_sender{}};
+auto tsd1 = time_single_sender<int, std::exception_ptr>{time_single_sender{}};
 ```
 
 ## put it all together with some algorithms

TARGETS

@@ -7,11 +7,11 @@ cpp_library(
         "include/executor.h",
         "include/extension_points.h",
         "include/flowsingle.h",
-        "include/flowsingledeferred.h",
+        "include/flowsinglesender.h",
         "include/none.h",
         "include/piping.h",
         "include/single.h",
-        "include/singledeferred.h",
+        "include/singlesender.h",
         "include/traits.h",
         "include/trampoline.h",
     ],

benchmarks/PushmiBenchmarks.cpp

@@ -11,7 +11,7 @@
 
 #include "pushmi/none.h"
 #include "pushmi/flow_single.h"
-#include "pushmi/flow_single_deferred.h"
+#include "pushmi/flow_single_sender.h"
 #include "pushmi/entangle.h"
 
 #include "pool.h"

buildSingleHeader.cmake

@@ -34,20 +34,21 @@ set(header_files
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/boosters.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/piping.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/none.h"
-    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/deferred.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/sender.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single.h"
-    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single_deferred.h"
-    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_deferred.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single_sender.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/executor.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single.h"
-    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_deferred.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h"
-    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_deferred.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/async.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/from.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h"

examples/include/bulk.h

@@ -5,7 +5,7 @@
 // This source code is licensed under the MIT license found in the
 // LICENSE file in the root directory of this source tree.
 
-#include <pushmi/single_deferred.h>
+#include <pushmi/single_sender.h>
 
 namespace pushmi {
 
@@ -21,7 +21,7 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn {
       IF&& initFunc,
       RS&& selector) const {
     return [func, sb, se, driver, initFunc, selector](auto in){
-      return make_single_deferred(
+      return make_single_sender(
         [in, func, sb, se, driver, initFunc, selector](auto out) mutable {
           submit(in, make_single(std::move(out),
               [func, sb, se, driver, initFunc, selector](auto& out, auto input) {

examples/include/no_fail.h

@@ -4,7 +4,7 @@
 // This source code is licensed under the MIT license found in the
 // LICENSE file in the root directory of this source tree.
 
-#include <pushmi/single_deferred.h>
+#include <pushmi/single_sender.h>
 #include <pushmi/o/submit.h>
 
 namespace pushmi {
@@ -33,7 +33,7 @@ struct no_fail_fn {
     PUSHMI_TEMPLATE(class In)
       (requires Sender<In>)
     auto operator()(In in) const {
-      return ::pushmi::detail::deferred_from(
+      return ::pushmi::detail::sender_from(
         std::move(in),
         ::pushmi::detail::submit_transform_out<In>(out_impl<In>{})
       );

examples/include/pool.h

@@ -45,7 +45,7 @@ class pool {
 
   inline auto executor() {
     auto exec = execution::require(p.executor(), execution::never_blocking, execution::oneway);
-    return MAKE(time_single_deferred)(__pool_submit<decltype(exec)>{exec});
+    return MAKE(time_single_sender)(__pool_submit<decltype(exec)>{exec});
   }
 
   inline void stop() {p.stop();}

examples/set_done/set_done_2.cpp

@@ -14,7 +14,7 @@ using namespace pushmi::aliases;
 const bool setting_exists = false;
 
 auto get_setting() {
-  return mi::make_single_deferred(
+  return mi::make_single_sender(
     [](auto out){
       if(setting_exists) {
         op::just(42) | op::submit(out);
@@ -31,7 +31,7 @@ auto println = [](auto v){std::cout << v << std::endl;};
 template<class T, class E = std::exception_ptr>
 auto concat =
   [](auto in){
-    return mi::make_single_deferred(
+    return mi::make_single_sender(
       [in](auto out) mutable {
         ::pushmi::submit(in, mi::make_single(out,
         [](auto out, auto v){
@@ -54,9 +54,9 @@ int main()
     op::just(42) |
       op::transform([](int i) {
         if (i < 42) {
-          return mi::any_single_deferred<std::string>{op::empty<std::string>()};
+          return mi::any_single_sender<std::string>{op::empty<std::string>()};
         }
-        return mi::any_single_deferred<std::string>{op::just(std::to_string(i))};
+        return mi::any_single_sender<std::string>{op::just(std::to_string(i))};
       }) |
       concat<std::string> |
       op::submit(println);

examples/set_error/set_error_2.cpp

@@ -16,7 +16,7 @@ using namespace pushmi::aliases;
 template<class T, class E = std::exception_ptr>
 auto concat =
   [](auto in){
-    return mi::make_single_deferred(
+    return mi::make_single_sender(
       [in](auto out) mutable {
         ::pushmi::submit(in, mi::make_single(out,
         [](auto out, auto v){
@@ -63,7 +63,7 @@ int main()
 
   op::just(42) |
     op::transform([](auto v) {
-      using r_t = mi::any_single_deferred<int>;
+      using r_t = mi::any_single_sender<int>;
       if (v < 40) {
         return r_t{op::error<int>(std::exception_ptr{})};
       } else {

examples/then_execute/then_execute_2.cpp

@@ -13,8 +13,8 @@
 
 #include <pool.h>
 
-#include <pushmi/deferred.h>
-#include <pushmi/single_deferred.h>
+#include <pushmi/sender.h>
+#include <pushmi/single_sender.h>
 #include <pushmi/o/just.h>
 #include <pushmi/o/via.h>
 #include <pushmi/o/transform.h>

examples/twoway_execute/twoway_execute_2.cpp

@@ -12,8 +12,8 @@
 
 #include <pool.h>
 
-#include <pushmi/deferred.h>
-#include <pushmi/single_deferred.h>
+#include <pushmi/sender.h>
+#include <pushmi/single_sender.h>
 #include <pushmi/o/transform.h>
 
 using namespace pushmi::aliases;