Commit 39d4be19 authored by Eric Niebler's avatar Eric Niebler Committed by Facebook Github Bot

Back-port pushmi from C++20+modules to standard C++14. (#16)

fbshipit-source-id: 2363aa0c02923504fcc36d927604a27a2f0558ba
parent ed3ab09c
...@@ -6,35 +6,52 @@ project(pushmi-project CXX) ...@@ -6,35 +6,52 @@ project(pushmi-project CXX)
FIND_PACKAGE (Threads) FIND_PACKAGE (Threads)
add_library(pushmi INTERFACE) add_library(pushmi INTERFACE)
target_include_directories(pushmi INTERFACE target_include_directories(pushmi INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include> $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/networking-ts-impl/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include> $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/executors-impl/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/futures-impl/future-executor-interaction/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/networking-ts-impl/include>
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/Catch2/single_include> $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/external/Catch2/single_include>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include> $<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/net>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/meta> $<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/meta>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/executors>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/futures>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/net>
$<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/Catch2>) $<INSTALL_INTERFACE:$<INSTALL_PREFIX>/include/Catch2>)
#target_compile_features(pushmi INTERFACE cxx_std_17) target_compile_features(pushmi INTERFACE cxx_std_14)
target_compile_options(pushmi INTERFACE target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:GNU>:-std=c++2a> #$<$<CXX_COMPILER_ID:GNU>:-std=c++2a>
$<$<CXX_COMPILER_ID:GNU>:-fconcepts>) #$<$<CXX_COMPILER_ID:GNU>:-fconcepts>
$<$<CXX_COMPILER_ID:GNU>:-ftemplate-backtrace-limit=0>)
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( install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/ DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/
DESTINATION include) DESTINATION include)
install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/networking-ts-impl/include/
DESTINATION include/net)
install( install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include/ DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include/
DESTINATION include/meta) DESTINATION include/meta)
install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/executors-impl/include/
DESTINATION include/executors)
install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/futures-impl/future-executor-interaction/include/
DESTINATION include/futures)
install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/networking-ts-impl/include/
DESTINATION include/net)
install( install(
DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/Catch2/ DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/external/Catch2/
DESTINATION include/Catch2) DESTINATION include/Catch2)
install(TARGETS pushmi EXPORT pushmi-project) install(TARGETS pushmi EXPORT pushmi-project)
install(EXPORT pushmi-project DESTINATION pushmi-project) install(EXPORT pushmi-project DESTINATION pushmi-project)
add_subdirectory(examples)
enable_testing() enable_testing()
include(CTest) include(CTest)
add_subdirectory(test) add_subdirectory(test)
function(BuildSingleHeader HeaderName)
set(header_files ${ARGN})
# write out single-header
file(READ "${CMAKE_CURRENT_SOURCE_DIR}/include/${HeaderName}-single-header.h" header)
string(APPEND incls "${header}")
foreach(f ${header_files})
message("processing ${f}")
file(READ ${f} contents)
string(REGEX REPLACE "(([\t ]*#[\t ]*pragma[\t ]+once)|([\t ]*#[\t ]*include))" "//\\1" filtered "${contents}")
string(APPEND incls "${filtered}")
endforeach()
file(READ "${CMAKE_CURRENT_SOURCE_DIR}/include/${HeaderName}-single-footer.h" footer)
string(APPEND incls "${footer}")
file(WRITE ${CMAKE_CURRENT_SOURCE_DIR}/include/${HeaderName}.h "${incls}")
endfunction()
set(header_files
# keep in inclusion order
"${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include/meta/meta_fwd.hpp"
"${CMAKE_CURRENT_SOURCE_DIR}/external/meta/include/meta/meta.hpp"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/detail/if_constexpr.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/detail/concept_def.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/traits.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/detail/functional.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/detail/opt.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/forwards.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/extension_points.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/properties.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/concepts.h"
"${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/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/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/trampoline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/on.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/tap.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/transform.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/via.h"
)
BuildSingleHeader("pushmi" ${header_files})
\ No newline at end of file
add_library(examples INTERFACE)
target_include_directories(examples INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>)
add_subdirectory(twoway_execute)
add_subdirectory(then_execute)
add_subdirectory(composition)
add_subdirectory(for_each)
add_subdirectory(reduce)
add_executable(composition_2 composition_2.cpp)
target_link_libraries(composition_2
pushmi
examples
Threads::Threads)
add_executable(composition_3 composition_3.cpp)
target_link_libraries(composition_3
pushmi
examples
Threads::Threads)
add_executable(composition_4 composition_4.cpp)
target_link_libraries(composition_4
pushmi
examples
Threads::Threads)
\ No newline at end of file
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <pool.h>
#include <pushmi/o/transform.h>
#include <pushmi/o/via.h>
using namespace pushmi::aliases;
struct f_t {};
f_t f(){
return {};
}
struct g_t {};
g_t g(f_t){
return {};
}
// these expressions are read backward, bottom-right to top-left
template<class CPUExecutor, class IOExecutor>
void lisp(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::transform([](auto){ return f(); })(
cpu)));
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::via([cpu]{return cpu;})(
op::transform([](auto){ return f(); })(
cpu))));
// f on io - g on cpu
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::via([cpu]{return cpu;})(
op::transform([](auto){ return f(); })(
io))));
}
template<class CPUExecutor, class IOExecutor>
void sugar(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
cpu |
op::transform([](auto){ return f(); }) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
cpu |
op::transform([](auto){ return f(); }) |
op::via([cpu]{return cpu;}) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
// f on io - g on cpu
io |
op::transform([](auto){ return f(); }) |
op::via([cpu]{return cpu;}) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
}
template<class CPUExecutor, class IOExecutor>
void pipe(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
mi::pipe(
cpu,
op::transform([](auto){ return f(); }),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
mi::pipe(
cpu,
op::transform([](auto){ return f(); }),
op::via([cpu]{return cpu;}),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
// f on io - g on cpu
mi::pipe(
io,
op::transform([](auto){ return f(); }),
op::via([cpu]{return cpu;}),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
}
int main()
{
mi::pool cpuPool{std::max(1u,std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u,std::thread::hardware_concurrency())};
lisp(cpuPool.executor(), ioPool.executor());
sugar(cpuPool.executor(), ioPool.executor());
pipe(cpuPool.executor(), ioPool.executor());
ioPool.wait();
cpuPool.wait();
std::cout << "OK" << std::endl;
}
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <defer.h>
#include <share.h>
#include <pushmi/o/just.h>
#include <pushmi/o/tap.h>
using namespace pushmi::aliases;
// three models of submission deferral
// (none of these use an executor, they are all running
// synchronously on the main thread)
// this constructs eagerly and submits just() lazily
auto defer_execution() {
printf("construct just\n");
return op::just(42) |
op::tap([](int v){ printf("just - %d\n", v); });
}
// this constructs defer() eagerly, constructs just() and submits just() lazily
auto defer_construction() {
return op::defer([]{
return defer_execution();
});
}
// this constructs defer(), constructs just() and submits just() eagerly
auto eager_execution() {
return defer_execution() | op::share<int>();
}
int main()
{
printf("\ncall defer_execution\n");
auto de = defer_execution();
printf("submit defer_execution\n");
de | op::submit();
// call defer_execution
// construct just
// submit defer_execution
// just - 42
printf("\ncall defer_construction\n");
auto dc = defer_construction();
printf("submit defer_construction\n");
dc | op::submit();
// call defer_construction
// submit defer_construction
// construct just
// just - 42
printf("\ncall eager_execution\n");
auto ee = eager_execution();
printf("submit eager_execution\n");
ee | op::submit();
// call eager_execution
// construct just
// just - 42
// submit eager_execution
std::cout << "OK" << std::endl;
// OK
}
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <pool.h>
#include <request_via.h>
#include <pushmi/o/tap.h>
#include <pushmi/o/transform.h>
using namespace pushmi::aliases;
template<class Io>
auto io_operation(Io io) {
return io |
op::transform([](auto){ return 42; }) |
op::tap([](int v){ printf("io pool producing, %d\n", v); }) |
op::request_via();
}
int main()
{
mi::pool cpuPool{std::max(1u,std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u,std::thread::hardware_concurrency())};
auto io = ioPool.executor();
auto cpu = cpuPool.executor();
io_operation(io).via([cpu]{ return cpu; }) |
op::tap([](int v){ printf("cpu pool processing, %d\n", v); }) |
op::submit();
// when the caller is not going to process the result (only side-effect matters)
// or the caller is just going to push the result into a queue.
// provide a way to skip the transition to a different executor and make it
// stand out so that it has to be justified in code reviews.
mi::via_cast<mi::is_sender<>>(io_operation(io)) | op::submit();
ioPool.wait();
cpuPool.wait();
std::cout << "OK" << std::endl;
}
add_executable(for_each_2 for_each_2.cpp)
target_link_libraries(for_each_2
pushmi
examples
Threads::Threads)
add_executable(for_each_3 for_each_3.cpp)
target_link_libraries(for_each_3
pushmi
examples
Threads::Threads)
\ No newline at end of file
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <pool.h>
#include <for_each.h>
using namespace pushmi::aliases;
template<class Executor, class Allocator = std::allocator<char>>
auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
return [e, a]<class IF, class RS, class Input, class F, class ShapeBegin, class ShapeEnd, class Out>(
IF init,
RS selector,
Input input,
F&& func,
ShapeBegin sb,
ShapeEnd se,
Out out) {
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> allocState(a);
auto shared_state = std::allocate_shared<
std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to first exception)
>>(allocState, std::exception_ptr{}, std::move(out), std::move(selector), std::move(func), init(std::move(input)), 1, 0);
e | op::submit([e, sb, se, shared_state](auto ){
auto stepDone = [](auto shared_state){
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch(...) {
mi::set_error(std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se; ++idx, ++std::get<5>(*shared_state)){
e | op::submit([shared_state, idx, stepDone](auto ex){
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3>(*shared_state)(step, idx);
} while(!std::get<4>(*shared_state).compare_exchange_strong(old, step));
} catch(...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
}
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()]() mutable {
mi::set_error(out, ep);
});
}
};
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
std::vector<int> vec(10);
mi::for_each(naive_executor_bulk_target(p.executor()), vec.begin(), vec.end(), [](int& x){
x = 42;
});
assert(std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
std::cout << "OK" << std::endl;
p.wait();
}
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <for_each.h>
using namespace pushmi::aliases;
auto inline_bulk_target() {
return []<class IF, class RS, class Input, class F, class ShapeBegin, class ShapeEnd, class Out>(
IF init,
RS selector,
Input input,
F&& func,
ShapeBegin sb,
ShapeEnd se,
Out out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx){
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch(...) {
mi::set_error(out, std::current_exception());
}
};
}
int main()
{
std::vector<int> vec(10);
mi::for_each(inline_bulk_target(), vec.begin(), vec.end(), [](int& x){
x = 42;
});
assert(std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
std::cout << "OK" << std::endl;
}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// 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>
#if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_
#define MAKE_(...) {__VA_ARGS__}
#else
#define MAKE(x) make_ ## x
#endif
namespace pushmi {
namespace operators {
template<class F, class ShapeBegin, class ShapeEnd, class Target, class IF, class RS>
auto bulk(
F&& func,
ShapeBegin sb,
ShapeEnd se,
Target&& driver,
IF&& initFunc,
RS&& selector) {
return [func, sb, se, driver, initFunc, selector](auto in){
return MAKE(single_deferred)([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){
driver(initFunc, selector, std::move(input), func, sb, se, std::move(out));
}
));
});
};
}
} // namespace operators
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// 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.h>
#include <pushmi/o/submit.h>
#include <pushmi/o/extension_operators.h>
namespace pushmi {
namespace operators {
PUSHMI_TEMPLATE(class F)
(requires Invocable<F>)
auto defer(F f) {
return make_single_deferred(
constrain(lazy::Receiver<_1>,
[f = std::move(f)](auto out) mutable {
auto sender = f();
PUSHMI_IF_CONSTEXPR( ((bool)TimeSender<decltype(sender)>) (
::pushmi::submit(sender, ::pushmi::now(id(sender)), std::move(out));
) else (
::pushmi::submit(sender, std::move(out));
));
}
)
);
}
} // namespace operators
} // namespace pushmi
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <bulk.h>
#include <pushmi/o/submit.h>
#include <pushmi/o/just.h>
namespace pushmi {
template<class ExecutionPolicy, class RandomAccessIterator, class Function>
void for_each(
ExecutionPolicy&& policy,
RandomAccessIterator begin,
RandomAccessIterator end,
Function f)
{
operators::just(0) |
operators::bulk(
[f](auto& acc, auto cursor){ f(*cursor); },
begin,
end,
policy,
[](auto&& args){ return args; },
[](auto&& acc){ return 0; }) |
operators::blocking_submit();
}
} // namespace pushmi
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <experimental/thread_pool>
#include <pushmi/executor.h>
#include <pushmi/trampoline.h>
namespace pushmi {
using std::experimental::static_thread_pool;
namespace execution = std::experimental::execution;
template<class Executor>
struct __pool_submit {
using e_t = Executor;
e_t e;
explicit __pool_submit(e_t e) : e(std::move(e)) {}
PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out>)
void operator()(TP at, Out out) const {
e.execute([e = this->e, at = std::move(at), out = std::move(out)]() mutable {
auto tr = trampoline();
::pushmi::submit(tr, std::move(at), std::move(out));
});
}
};
class pool {
static_thread_pool p;
public:
inline explicit pool(std::size_t threads) : p(threads) {}
inline auto executor() {
auto exec = execution::require(p.executor(), execution::never_blocking, execution::oneway);
return make_time_single_deferred(__pool_submit<decltype(exec)>{exec});
}
inline void stop() {p.stop();}
inline void wait() {p.wait();}
};
} // namespace pushmi
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <bulk.h>
#include <pushmi/o/submit.h>
#include <pushmi/o/just.h>
namespace pushmi {
template<class ExecutionPolicy, class ForwardIt, class T, class BinaryOp>
T reduce(
ExecutionPolicy&& policy,
ForwardIt begin,
ForwardIt end,
T init,
BinaryOp binary_op){
return operators::just(std::move(init)) |
operators::bulk(
[binary_op](auto& acc, auto cursor){ acc = binary_op(acc, *cursor); },
begin,
end,
policy,
[](auto&& args){ return args; },
[](auto&& acc){ return acc; }) |
operators::get<T>;
}
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// 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.h>
#include <pushmi/o/submit.h>
#include <pushmi/o/extension_operators.h>
#include <pushmi/o/via.h>
namespace pushmi {
namespace detail {
struct request_via_fn {
template<typename In>
struct semisender {
In in;
template<class... AN>
auto via(AN&&... an) {
return in | ::pushmi::operators::via((AN&&) an...);
}
};
auto operator()() const;
};
auto request_via_fn::operator()() const {
return constrain(lazy::Sender<_1>, [](auto in) {
using In = decltype(in);
return semisender<In>{in};
});
}
} // namespace detail
namespace operators {
PUSHMI_INLINE_VAR constexpr detail::request_via_fn request_via{};
} // namespace operators
PUSHMI_TEMPLATE(class To, class In)
(requires Same<To, is_sender<>> && Sender<_1>)
auto via_cast(In in) {
return in;
}
PUSHMI_TEMPLATE(class To, class In)
(requires Same<To, is_sender<>>)
auto via_cast(detail::request_via_fn::semisender<In> ss) {
return ss.in;
}
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// 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.h>
#include <pushmi/o/submit.h>
#include <pushmi/o/extension_operators.h>
#include <subject.h>
namespace pushmi {
namespace detail {
template<class T>
struct share_fn {
auto operator()() const;
};
template<class T>
auto share_fn<T>::operator()() const {
return constrain(lazy::Sender<_1>, [](auto in) {
using In = decltype(in);
subject<T, properties_t<In>> sub;
PUSHMI_IF_CONSTEXPR( ((bool)TimeSender<In>) (
::pushmi::submit(in, ::pushmi::now(id(in)), sub.receiver());
) else (
::pushmi::submit(id(in), sub.receiver());
));
return sub;
});
}
} // namespace detail
namespace operators {
template<class T>
PUSHMI_INLINE_VAR constexpr detail::share_fn<T> share{};
} // namespace operators
} // namespace pushmi
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <pushmi/time_single_deferred.h>
namespace pushmi {
template<class... TN>
struct subject;
template<class T, class PS>
struct subject<T, PS> {
using properties = property_set_insert_t<property_set<is_sender<>, is_single<>>, PS>;
struct subject_shared {
bool done_ = false;
pushmi::detail::opt<T> t_;
std::exception_ptr ep_;
std::vector<any_single<T>> receivers_;
std::mutex lock_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
void submit(Out out) {
std::unique_lock<std::mutex> guard(lock_);
if (ep_) {::pushmi::set_error(out, ep_); return;}
if (!!t_) {::pushmi::set_value(out, (const T&)t_); return;}
if (done_) {::pushmi::set_done(out); return;}
receivers_.push_back(any_single<T>{out});
}
PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
void value(V&& v) {
std::unique_lock<std::mutex> guard(lock_);
t_ = detail::as_const(v);
for (auto& out : receivers_) {::pushmi::set_value(out, (V&&) v);}
receivers_.clear();
}
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>)
void error(E e) noexcept {
std::unique_lock<std::mutex> guard(lock_);
ep_ = e;
for (auto& out : receivers_) {::pushmi::set_error(out, std::move(e));}
receivers_.clear();
}
void done() {
std::unique_lock<std::mutex> guard(lock_);
done_ = true;
for (auto& out : receivers_) {::pushmi::set_done(out);}
receivers_.clear();
}
};
// need a template overload of none/deferred and the rest that stores a 'ptr' with its own lifetime management
struct subject_receiver {
using properties = property_set_insert_t<property_set<is_receiver<>, is_single<>>, PS>;
std::shared_ptr<subject_shared> s;
PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
void value(V&& v) {
s->value((V&&) v);
}
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>)
void error(E e) noexcept {
s->error(std::move(e));
}
void done() {
s->done();
}
};
std::shared_ptr<subject_shared> s = std::make_shared<subject_shared>();
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
void submit(Out out) {
s->submit(std::move(out));
}
auto receiver() {
return detail::out_from_fn<subject>{}(subject_receiver{s});
}
};
} // namespace pushmi
# things I learned from std::for_each and std::reduce
I wrote an operator called `bulk()` and implemented for_each and reduce in terms of it. I departed from the `bulk_execute()` signature and tried to model the reduce signature on my `bulk` operator. I am not satisfied with the result and would need to invest more to get an abstraction for bulk that I was confident was minimal and efficient.
# Background
## bulk_execute
The `bulk_execute()` function is intended to be an abstraction that allows efficient implementation of the parallel std algorithms on both CPU and GPU executors.
```cpp
template<class Function, class SharedFactory>
void bulk_execute(Function&& f, size_t n, SharedFactory&& sf) const;
```
A sequential implementation might look like:
```cpp
template<class Function, class SharedFactory>
void bulk_execute(Function&& f, size_t n, SharedFactory&& sf)
{
auto state = sf();
for(size_t idx = 0; idx < n; ++idx) {
f(state, idx);
}
}
```
The `Function f` already appears to be similar to the accumulate function passed to reduce. It takes the shared state and the index indicating the current value. The SharedFactory is very similar to the initialValue parameter to reduce. The Shape parameter is very similar to the Range parameter to reduce. These similarities motivated me to modify the signature to more explicitly match the reduce pattern.
## bulk operator
```cpp
template<class F, class ShapeBegin, class ShapeEnd, class Target, class IF, class RS>
auto bulk(
F&& func,
ShapeBegin sb,
ShapeEnd se,
Target&& driver,
IF&& initFunc,
RS&& selector);
```
The `bulk` function packages the parameters and returns an adapter function.
> A Sender is an object with a `submit()` method
> An Adapter is a function that takes a Sender and returns a Sender. Adapters are used for composition.
When called, the Adapter from `bulk()` will package the Adapter parameter with the original parameters and return a Sender.
> A Receiver is an object that has methods like `value()`, `error()` and `done()`. A Receiver is like a Promise.
The `submit()` method takes a Receiver. When called, the Sender from the bulk Adapter will create a Receiver with the original parameters to `bulk()` and the Receiver parameter. This new Receiver will be passed to `submit()` on the Sender that the bulk Adapter stored in this bulk Sender.
When called, the `value()` method on the bulk Receiver will pass all the packaged parameters to the Target.
> A Target is a function that orchestrates the bulk operation using the parameters. There would be different Target implementations for device, sequential, concurrent execution patterns.
A Target implementation might look like:
```cpp
template<class IF, class RS, class Input, class F, class ShapeBegin, class ShapeEnd, class Out>
void inline_driver(
IF init,
RS selector,
Input input,
F&& func,
ShapeBegin sb,
ShapeEnd se,
Out out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx){
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch(...) {
mi::set_error(out, std::current_exception());
}
};
```
> ways to improve bulk:
> - merge ShapeBegin and ShapeEnd into a Range.
> - pass out to selector so that it can deliver an error or a success.
> - initFunc multiple times to have context local state that does not need locking or CAS loop.
> - compose the driver implementations from operators rather than each having a bespoke implementation
# for_each
implementing for_each was straight-forward with the interface.
```cpp
template<class ExecutionPolicy, class RandomAccessIterator, class Function>
void for_each(
ExecutionPolicy&& policy,
RandomAccessIterator begin,
RandomAccessIterator end,
Function f)
{
operators::just(0) |
operators::bulk(
[f](auto& acc, auto cursor){ f(*cursor); },
begin,
end,
policy,
[](auto&& args){ return args; },
[](auto&& acc){ return 0; }) |
operators::blocking_submit();
}
```
The oddity is that bulk is expecting a shared state value and a value as input and a value result. Since for_each does not have shared state, this is overhead that becomes obvious and disturbing when looking at the naive concurrent driver in the code (there is a CAS loop around the call to the state update function even though the state is not updated here).
# reduce
implementing reduce took more effort and some of the code in the drivers and parameters to the driver were modified to get reduce working.
```cpp
template<class ExecutionPolicy, class ForwardIt, class T, class BinaryOp>
T reduce(
ExecutionPolicy&& policy,
ForwardIt begin,
ForwardIt end,
T init,
BinaryOp binary_op){
return operators::just(std::move(init)) |
operators::bulk(
[binary_op](auto& acc, auto cursor){ acc = binary_op(acc, *cursor); },
begin,
end,
policy,
[](auto&& args){ return args; },
[](auto&& acc){ return acc; }) |
operators::get<T>;
}
```
Based on examples that I have been shown, the existing bulk_execute would expect the bulk_execute caller to provide the synchronization for the shared state. In the case of reduce it is important to synchronize when the execution is concurrent and equally important not to synchronize when it is not concurrent. Using if constexpr to implement reduce with and without sync in the parameters to bulk would add a lot of unsafe bespoke work and complication into every algorithm using bulk. In this bulk design the driver owns synchronization instead.
> NOTE - in any case, if a high-level async library design requires manual lock or lock-free primitive usage for correct behavior, then the design needs to be changed.
I am dissatisfied with the expected perf results from the naive concurrent driver (expectation from looking at the code, will need to measure). For instance, here is the CAS loop over the accumulator function from the naive concurrent driver:
```cpp
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3>(*shared_state)(step, idx);
} while(!std::get<4>(*shared_state).compare_exchange_strong(old, step));
```
This is due to having a single shared_state being shared concurrently. I would much prefer having multiple states that are never used concurrently and then composing them all into one final result.
> creating factor * hardware_concurrency() number of states would allow user controlled granularity (factor) for work stealing. each state would only be used from one `hardware_concurrency` context and thus would have no synchronization when it was modified.
# static_thread_pool
this bonus section is to mention the bulk_execute implementation in the static_thread_pool. The static thread pool is a cool piece of tech. in the bulk_execute method I had two observations.
1. every index in the range from 0-N is allocated as a task node
2. this list of nodes is built locally and then inserted in one lock operation
For #1, I expect that there is a desire to reduce the number of task nodes allocated in bulk.
There are multiple ways to achieve #2 on P1055.
one way to achieve this is to add a type that is an executor but just accumulates a local queue. usage would be similar to..
```cpp
auto pool = thread_pool();
auto e = pool.bulk_executor();
my_bulk_generator(e, . . .); // lots of calls to submit
pool.bulk_enqueue(e);
```
# ExecutionPolicy
In building these algorithms I observed that the parallel std algorithms do not really depend on executor, they depend on ExecutionPolicy. GPU and CPU can have different execution policies and it does not affect the implementation or expression of the parallel algorithms (rather than passing an executor around pass an ExecutionPolicy).
add_executable(reduce_2 reduce_2.cpp)
target_link_libraries(reduce_2
pushmi
examples
Threads::Threads)
add_executable(reduce_3 reduce_3.cpp)
target_link_libraries(reduce_3
pushmi
examples
Threads::Threads)
\ No newline at end of file
#include <vector>
#include <algorithm>
#include <numeric>
#include <cassert>
#include <iostream>
#include <exception>
#include <pool.h>
#include <reduce.h>
using namespace pushmi::aliases;
template<class Executor, class Allocator = std::allocator<char>>
auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
return [e, a]<class IF, class RS, class Input, class F, class ShapeBegin, class ShapeEnd, class Out>(
IF init,
RS selector,
Input input,
F&& func,
ShapeBegin sb,
ShapeEnd se,
Out out) {
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> allocState(a);
auto shared_state = std::allocate_shared<
std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to first exception)
>>(allocState, std::exception_ptr{}, std::move(out), std::move(selector), std::move(func), init(std::move(input)), 1, 0);
e | op::submit([e, sb, se, shared_state](auto ){
auto stepDone = [](auto shared_state){
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch(...) {
mi::set_error(std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se; ++idx, ++std::get<5>(*shared_state)){
e | op::submit([shared_state, idx, stepDone](auto ex){
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3>(*shared_state)(step, idx);
} while(!std::get<4>(*shared_state).compare_exchange_strong(old, step));
} catch(...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
}
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()]() mutable {
mi::set_error(out, ep);
});
}
};
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
std::vector<int> vec(10);
std::fill(vec.begin(), vec.end(), 4);
auto fortyTwo = mi::reduce(naive_executor_bulk_target(p.executor()), vec.begin(), vec.end(), 2, std::plus<>{});
assert(std::accumulate(vec.begin(), vec.end(), 2) == fortyTwo);
std::cout << "OK" << std::endl;
p.wait();
}
#include <vector>
#include <numeric>
#include <cassert>
#include <iostream>
#include <exception>
#include <reduce.h>
using namespace pushmi::aliases;
auto inline_bulk_target() {
return []<class IF, class RS, class Input, class F, class ShapeBegin, class ShapeEnd, class Out>(
IF init,
RS selector,
Input input,
F&& func,
ShapeBegin sb,
ShapeEnd se,
Out out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx){
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch(...) {
mi::set_error(out, std::current_exception());
}
};
}
int main()
{
std::vector<int> vec(10);
std::fill(vec.begin(), vec.end(), 4);
auto fortyTwo = mi::reduce(inline_bulk_target(), vec.begin(), vec.end(), 2, std::plus<>{});
assert(std::accumulate(vec.begin(), vec.end(), 2) == fortyTwo);
std::cout << "OK" << std::endl;
}
\ No newline at end of file
add_executable(then_execute_2 then_execute_2.cpp)
target_link_libraries(then_execute_2
pushmi
examples
Threads::Threads)
\ No newline at end of file
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <memory>
#include <utility>
#include <atomic>
#include <thread>
#include <functional>
#include <futures.h>
#include <futures_static_thread_pool.h>
#include <pool.h>
#include <request_via.h>
#include <share.h>
#include <pushmi/deferred.h>
#include <pushmi/single_deferred.h>
#include <pushmi/o/just.h>
#include <pushmi/o/transform.h>
using namespace pushmi::aliases;
struct inline_executor
{
public:
friend bool operator==(
const inline_executor&, const inline_executor&) noexcept { return true; }
friend bool operator!=(
const inline_executor&, const inline_executor&) noexcept { return false; }
template<class Function>
void execute(Function f) const noexcept { f(); }
constexpr bool query(std::experimental::execution::oneway_t) { return true; }
constexpr bool query(std::experimental::execution::twoway_t) { return false; }
constexpr bool query(std::experimental::execution::single_t) { return true; }
};
namespace p1054 {
// A promise refers to a promise and is associated with a future,
// either through type-erasure or through construction of an
// underlying promise with an overload of make_promise_contract().
// make_promise_contract() cannot be written to produce a lazy future.
// the promise has to exist prior to .then() getting a continuation.
// there must be a shared allocation to connect the promise and future.
template <class T, class Executor>
std::pair<std::experimental::standard_promise<T>, std::experimental::standard_future<T, std::decay_t<Executor>>>
make_promise_contract(const Executor& e) {
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
}
template<class Executor, class Function, class Future>
std::experimental::standard_future<
std::result_of_t<Function(std::decay_t<typename std::decay_t<Future>::value_type>&&)>, std::decay_t<Executor>>
then_execute(Executor&& e, Function&& f, Future&& pred) {
using V = std::decay_t<typename std::decay_t<Future>::value_type>;
using T = std::result_of_t<Function(V&&)>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
((Future&&)pred).then([e, p, f](V v) mutable {
e.execute([p, f, v]() mutable {
p.set_value(f(v));
});
return 0;
});
return r;
}
}
namespace p1055 {
template<class Executor, class Function, class Future>
auto then_execute(Executor&& e, Function&& f, Future&& pred) {
return pred | op::via([e](){return e;}) | op::transform([f](auto v){return f(v);});
}
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
std::experimental::futures_static_thread_pool sp{std::max(1u,std::thread::hardware_concurrency())};
auto pc = p1054::make_promise_contract<int>(inline_executor{});
auto& pr = std::get<0>(pc);
auto& r = std::get<1>(pc);
auto f = p1054::then_execute(sp.executor(), [](int v){return v*2;}, std::move(r));
pr.set_value(42);
f.get();
p1055::then_execute(p.executor(), [](int v){return v*2;}, op::just(21)) | op::get<int>;
sp.wait();
p.wait();
std::cout << "OK" << std::endl;
}
add_executable(twoway_execute_2 twoway_execute_2.cpp)
target_link_libraries(twoway_execute_2
pushmi
examples
Threads::Threads)
\ No newline at end of file
#include <vector>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <memory>
#include <utility>
#include <atomic>
#include <thread>
#include <functional>
#include <futures.h>
#include <pool.h>
#include <request_via.h>
#include <share.h>
#include <pushmi/deferred.h>
#include <pushmi/single_deferred.h>
#include <pushmi/o/transform.h>
using namespace pushmi::aliases;
namespace p1054 {
// A promise refers to a promise and is associated with a future,
// either through type-erasure or through construction of an
// underlying promise with an overload of make_promise_contract().
// make_promise_contract() cannot be written to produce a lazy future.
// the promise has to exist prior to .then() getting a continuation.
// there must be a shared allocation to connect the promise and future.
template <class T, class Executor>
std::pair<std::experimental::standard_promise<T>, std::experimental::standard_future<T, std::decay_t<Executor>>>
make_promise_contract(const Executor& e) {
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
}
template<class Executor, class Function>
std::experimental::standard_future<std::result_of_t<std::decay_t<Function>()>, std::decay_t<Executor>>
twoway_execute(Executor&& e, Function&& f) {
using T = std::result_of_t<std::decay_t<Function>()>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
e.execute([p, f]() mutable {p.set_value(f());});
return r;
}
}
namespace p1055 {
template<class Executor, class Function>
auto twoway_execute(Executor&& e, Function&& f) {
return e | op::transform([f](auto){return f();});
}
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
std::experimental::static_thread_pool sp{std::max(1u,std::thread::hardware_concurrency())};
p1054::twoway_execute(sp.executor(), [](){return 42;}).get();
p1055::twoway_execute(p.executor(), [](){return 42;}) | op::get<int>;
sp.wait();
p.wait();
std::cout << "OK" << std::endl;
}
#pragma once
#ifndef PUSHMI_SINGLE_HEADER
#define PUSHMI_SINGLE_HEADER
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <cstddef>
#include <cassert>
#include <cstdint>
#include <cstdio>
#include <chrono>
#include <exception>
#include <functional>
#include <algorithm>
#include <utility>
#include <memory>
#include <type_traits>
#include <initializer_list>
#include <thread>
#include <future>
#include <tuple>
#include <deque>
#if __cpp_lib_optional >= 201606
#include <optional>
#endif
This diff is collapsed.
...@@ -17,8 +17,8 @@ namespace pushmi { ...@@ -17,8 +17,8 @@ namespace pushmi {
template<class T> template<class T>
struct construct { struct construct {
template<class... AN> PUSHMI_TEMPLATE(class... AN)
requires Constructible<T, AN...> (requires Constructible<T, AN...>)
auto operator()(AN&&... an) const { auto operator()(AN&&... an) const {
return T{std::forward<AN>(an)...}; return T{std::forward<AN>(an)...};
} }
...@@ -28,23 +28,13 @@ template<template <class...> class T> ...@@ -28,23 +28,13 @@ template<template <class...> class T>
struct construct_deduced; struct construct_deduced;
template<> template<>
struct construct_deduced<none> { struct construct_deduced<none>;
template<class... AN>
auto operator()(AN&&... an) const -> decltype(none{(AN&&) an...}) {
return none{(AN&&) an...};
}
};
template<> template<>
struct construct_deduced<single> { struct construct_deduced<single>;
template<class... AN>
auto operator()(AN&&... an) const -> decltype(single{(AN&&) an...}) {
return single{(AN&&) an...};
}
};
template <template <class...> class T, class... AN> template <template <class...> class T, class... AN>
using deduced_type_t = std::invoke_result_t<construct_deduced<T>, AN...>; using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>;
struct ignoreVF { struct ignoreVF {
template <class V> template <class V>
...@@ -84,41 +74,44 @@ struct systemNowF { ...@@ -84,41 +74,44 @@ struct systemNowF {
}; };
struct passDVF { struct passDVF {
template <class V, Receiver Data> PUSHMI_TEMPLATE(class V, class Data)
requires requires(Data& out, V&& v) { (requires requires (
::pushmi::set_value(out, (V&&) v); ::pushmi::set_value(std::declval<Data&>(), std::declval<V>())
} ) && Receiver<Data>)
void operator()(Data& out, V&& v) const { void operator()(Data& out, V&& v) const {
::pushmi::set_value(out, (V&&) v); ::pushmi::set_value(out, (V&&) v);
} }
}; };
struct passDEF { struct passDEF {
template <class E, NoneReceiver<E> Data> PUSHMI_TEMPLATE(class E, class Data)
(requires NoneReceiver<Data, E>)
void operator()(Data& out, E e) const noexcept { void operator()(Data& out, E e) const noexcept {
::pushmi::set_error(out, e); ::pushmi::set_error(out, e);
} }
}; };
struct passDDF { struct passDDF {
template <Receiver Data> PUSHMI_TEMPLATE(class Data)
(requires Receiver<Data>)
void operator()(Data& out) const { void operator()(Data& out) const {
::pushmi::set_done(out); ::pushmi::set_done(out);
} }
}; };
struct passDStpF { struct passDStpF {
template <Receiver Data> PUSHMI_TEMPLATE(class Data)
(requires Receiver<Data>)
void operator()(Data& out) const { void operator()(Data& out) const {
::pushmi::set_stopping(out); ::pushmi::set_stopping(out);
} }
}; };
struct passDStrtF { struct passDStrtF {
template <class Up, Receiver Data> PUSHMI_TEMPLATE(class Up, class Data)
requires requires(Data& out, Up& up) { (requires requires (
::pushmi::set_starting(out, up); ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up&>())
} ) && Receiver<Data>)
void operator()(Data& out, Up& up) const { void operator()(Data& out, Up& up) const {
::pushmi::set_starting(out, up); ::pushmi::set_starting(out, up);
} }
...@@ -137,84 +130,119 @@ struct passDSF { ...@@ -137,84 +130,119 @@ struct passDSF {
}; };
struct passDNF { struct passDNF {
template <TimeSender Data> PUSHMI_TEMPLATE(class Data)
(requires TimeSender<Data>)
auto operator()(Data& in) const noexcept { auto operator()(Data& in) const noexcept {
return ::pushmi::now(in); return ::pushmi::now(in);
} }
}; };
// inspired by Ovrld - shown in a presentation by Nicolai Josuttis // inspired by Ovrld - shown in a presentation by Nicolai Josuttis
template <SemiMovable... Fns> #if __cpp_variadic_using >= 201611 && __cpp_concepts
struct overload : Fns... { template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... Fns>
constexpr overload() = default; requires sizeof...(Fns) > 0
constexpr explicit overload(Fns... fns) requires sizeof...(Fns) == 1 struct overload_fn : Fns... {
constexpr overload_fn() = default;
constexpr explicit overload_fn(Fns... fns) requires sizeof...(Fns) == 1
: Fns(std::move(fns))... {} : Fns(std::move(fns))... {}
constexpr overload(Fns... fns) requires sizeof...(Fns) > 1 constexpr overload_fn(Fns... fns) requires sizeof...(Fns) > 1
: Fns(std::move(fns))... {} : Fns(std::move(fns))... {}
using Fns::operator()...; using Fns::operator()...;
}; };
#else
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... Fns>
#if __cpp_concepts
requires sizeof...(Fns) > 0
#endif
struct overload_fn;
template <class Fn>
struct overload_fn<Fn> : Fn {
constexpr overload_fn() = default;
constexpr explicit overload_fn(Fn fn)
: Fn(std::move(fn)) {}
using Fn::operator();
};
template <class Fn, class... Fns>
struct overload_fn<Fn, Fns...> : Fn, overload_fn<Fns...> {
constexpr overload_fn() = default;
constexpr overload_fn(Fn fn, Fns... fns)
: Fn(std::move(fn)), overload_fn<Fns...>{std::move(fns)...} {}
using Fn::operator();
using overload_fn<Fns...>::operator();
};
#endif
template <class... F> template <class... Fns>
overload(F...) -> overload<F...>; auto overload(Fns... fns) -> overload_fn<Fns...> {
return overload_fn<Fns...>{std::move(fns)...};
}
template <class... Fns> template <class... Fns>
struct on_value : overload<Fns...> { struct on_value_fn : overload_fn<Fns...> {
constexpr on_value() = default; constexpr on_value_fn() = default;
using overload<Fns...>::overload; using overload_fn<Fns...>::overload_fn;
using Fns::operator()...;
}; };
template <class... F> template <class... Fns>
on_value(F...)->on_value<F...>; auto on_value(Fns... fns) -> on_value_fn<Fns...> {
return on_value_fn<Fns...>{std::move(fns)...};
}
template <class... Fns> template <class... Fns>
struct on_error : overload<Fns...> { struct on_error_fn : overload_fn<Fns...> {
constexpr on_error() = default; constexpr on_error_fn() = default;
using overload<Fns...>::overload; using overload_fn<Fns...>::overload_fn;
using Fns::operator()...;
}; };
template <class... F>
on_error(F...)->on_error<F...>;
template <class... Fns> template <class... Fns>
struct on_done : overload<Fns...> { auto on_error(Fns... fns) -> on_error_fn<Fns...> {
constexpr on_done() = default; return on_error_fn<Fns...>{std::move(fns)...};
using overload<Fns...>::overload; }
using Fns::operator()...;
template <class Fn>
struct on_done_fn : Fn {
constexpr on_done_fn() = default;
constexpr explicit on_done_fn(Fn fn) : Fn(std::move(fn)) {}
using Fn::operator();
}; };
template <class F> template <class Fn>
on_done(F)->on_done<F>; auto on_done(Fn fn) -> on_done_fn<Fn> {
return on_done_fn<Fn>{std::move(fn)};
}
template <class... Fns> template <class Fn>
struct on_stopping : overload<Fns...> { struct on_stopping_fn : Fn {
constexpr on_stopping() = default; constexpr on_stopping_fn() = default;
using overload<Fns...>::overload; constexpr explicit on_stopping_fn(Fn fn) : Fn(std::move(fn)) {}
using Fns::operator()...; using Fn::operator();
}; };
template <class F> template <class Fn>
on_stopping(F)->on_stopping<F>; auto on_stopping(Fn fn) -> on_stopping_fn<Fn> {
return on_stopping_fn<Fn>{std::move(fn)};
}
template <class... Fns> template <class... Fns>
struct on_starting : overload<Fns...> { struct on_starting_fn : overload_fn<Fns...> {
constexpr on_starting() = default; constexpr on_starting_fn() = default;
using overload<Fns...>::overload; using overload_fn<Fns...>::overload_fn;
using Fns::operator()...;
}; };
template <class... F> template <class... Fns>
on_starting(F...)->on_starting<F...>; auto on_starting(Fns... fns) -> on_starting_fn<Fns...> {
return on_starting_fn<Fns...>{std::move(fns)...};
}
template <class... Fns> template <class... Fns>
struct on_submit : overload<Fns...> { struct on_submit_fn : overload_fn<Fns...> {
constexpr on_submit() = default; constexpr on_submit_fn() = default;
using overload<Fns...>::overload; using overload_fn<Fns...>::overload_fn;
using Fns::operator()...;
}; };
template <class... F> template <class... Fns>
on_submit(F...)->on_submit<F...>; auto on_submit(Fns... fns) -> on_submit_fn<Fns...> {
return on_submit_fn<Fns...>{std::move(fns)...};
}
} // namespace pushmi } // namespace pushmi
...@@ -20,44 +20,63 @@ class deferred<detail::erase_deferred_t, E> { ...@@ -20,44 +20,63 @@ class deferred<detail::erase_deferred_t, E> {
template <class Wrapped> template <class Wrapped>
static constexpr bool insitu() { static constexpr bool insitu() {
return sizeof(Wrapped) <= sizeof(data::buffer_) && return sizeof(Wrapped) <= sizeof(data::buffer_) &&
std::is_nothrow_move_constructible_v<Wrapped>; std::is_nothrow_move_constructible<Wrapped>::value;
} }
struct vtable { struct vtable {
void (*op_)(data&, data*) = +[](data&, data*) {}; static void s_op(data&, data*) {}
void (*submit_)(data&, any_none<E>) = +[](data&, any_none<E>) {}; static void s_submit(data&, any_none<E>) {}
static constexpr vtable const noop_ = {}; void (*op_)(data&, data*) = s_op;
void (*submit_)(data&, any_none<E>) = s_submit;
static constexpr vtable const noop_{};
} const* vptr_ = &vtable::noop_; } const* vptr_ = &vtable::noop_;
template <class Wrapped, bool = insitu<Wrapped>()>
static constexpr vtable const vtable_v = {
+[](data& src, data* dst) {
if (dst)
dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_);
},
+[](data& src, any_none<E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
template <class T, class U = std::decay_t<T>> template <class T, class U = std::decay_t<T>>
using wrapped_t = using wrapped_t =
std::enable_if_t<!std::is_same_v<U, deferred>, U>; std::enable_if_t<!std::is_same<U, deferred>::value, U>;
public: public:
using sender_category = none_tag; using properties = property_set<is_sender<>, is_none<>>;
deferred() = default; deferred() = default;
deferred(deferred&& that) noexcept : deferred() { deferred(deferred&& that) noexcept : deferred() {
that.vptr_->op_(that.data_, &data_); that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_); std::swap(that.vptr_, vptr_);
} }
template <class Wrapped> PUSHMI_TEMPLATE(class Wrapped)
requires SenderTo<wrapped_t<Wrapped>, any_none<E>, none_tag> (requires SenderTo<wrapped_t<Wrapped>, any_none<E>, is_none<>>
explicit deferred(Wrapped obj) PUSHMI_BROKEN_SUBSUMPTION(&& !insitu<Wrapped>()))
: deferred() { explicit deferred(Wrapped obj) : deferred() {
if constexpr (insitu<Wrapped>()) struct s {
new (data_.buffer_) Wrapped(std::move(obj)); static void op(data& src, data* dst) {
else if (dst)
data_.pobj_ = new Wrapped(std::move(obj)); dst->pobj_ = std::exchange(src.pobj_, nullptr);
vptr_ = &vtable_v<Wrapped>; delete static_cast<Wrapped const*>(src.pobj_);
}
static void submit(data& src, any_none<E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
static const vtable vtbl{s::op, s::submit};
data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl;
}
PUSHMI_TEMPLATE(class Wrapped)
(requires SenderTo<wrapped_t<Wrapped>, any_none<E>, is_none<>>
&& insitu<Wrapped>())
explicit deferred(Wrapped obj) noexcept : deferred() {
struct s {
static void op(data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
static void submit(data& src, any_none<E> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
}
};
static const vtable vtbl{s::op, s::submit};
new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl;
} }
~deferred() { ~deferred() {
vptr_->op_(data_, nullptr); vptr_->op_(data_, nullptr);
...@@ -76,49 +95,32 @@ class deferred<detail::erase_deferred_t, E> { ...@@ -76,49 +95,32 @@ class deferred<detail::erase_deferred_t, E> {
template <class E> template <class E>
constexpr typename deferred<detail::erase_deferred_t, E>::vtable const constexpr typename deferred<detail::erase_deferred_t, E>::vtable const
deferred<detail::erase_deferred_t, E>::vtable::noop_; deferred<detail::erase_deferred_t, E>::vtable::noop_;
template <class E>
template <class Wrapped, bool Big>
constexpr typename deferred<detail::erase_deferred_t, E>::vtable const
deferred<detail::erase_deferred_t, E>::vtable_v;
template <class E>
template <class Wrapped>
constexpr typename deferred<detail::erase_deferred_t, E>::vtable const
deferred<detail::erase_deferred_t, E>::vtable_v<Wrapped, true> = {
+[](data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
},
+[](data& src, any_none<E> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
}
};
template <class SF> template <class SF>
class deferred<SF> { class deferred<SF> {
SF sf_; SF sf_{};
public: public:
using sender_category = none_tag; using properties = property_set<is_sender<>, is_none<>>;
constexpr deferred() = default; constexpr deferred() = default;
constexpr explicit deferred(SF sf) : sf_(std::move(sf)) {} constexpr explicit deferred(SF sf) : sf_(std::move(sf)) {}
template <Receiver<none_tag> Out> PUSHMI_TEMPLATE(class Out)
requires Invocable<SF&, Out> (requires Receiver<Out, is_none<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
sf_(std::move(out)); sf_(std::move(out));
} }
}; };
template <Sender<none_tag> Data, class DSF> template <PUSHMI_TYPE_CONSTRAINT(Sender<is_none<>>) Data, class DSF>
class deferred<Data, DSF> { class deferred<Data, DSF> {
Data data_; Data data_{};
DSF sf_; DSF sf_{};
static_assert(Sender<Data, is_none<>>, "The Data template parameter "
"must satisfy the Sender concept.");
public: public:
using sender_category = none_tag; using properties = property_set<is_sender<>, is_none<>>;
constexpr deferred() = default; constexpr deferred() = default;
constexpr explicit deferred(Data data) constexpr explicit deferred(Data data)
...@@ -126,35 +128,62 @@ class deferred<Data, DSF> { ...@@ -126,35 +128,62 @@ class deferred<Data, DSF> {
constexpr deferred(Data data, DSF sf) constexpr deferred(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
template <Receiver<none_tag> Out> PUSHMI_TEMPLATE(class Out)
requires Invocable<DSF&, Data&, Out> (requires Receiver<Out, is_none<>> && Invocable<DSF&, Data&, Out>)
void submit(Out out) { void submit(Out out) {
sf_(data_, std::move(out)); sf_(data_, std::move(out));
} }
}; };
////////////////////////////////////////////////////////////////////////////////
// make_deferred
inline auto make_deferred() -> deferred<ignoreSF> {
return deferred<ignoreSF>{};
}
template <class SF>
auto make_deferred(SF sf) -> deferred<SF> {
return deferred<SF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class Wrapped)
(requires Sender<Wrapped, is_none<>>)
auto make_deferred(Wrapped w) ->
deferred<detail::erase_deferred_t, std::exception_ptr> {
return deferred<detail::erase_deferred_t, std::exception_ptr>{std::move(w)};
}
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_none<>>)
auto make_deferred(Data data, DSF sf) -> deferred<Data, DSF> {
return deferred<Data, DSF>{std::move(data), std::move(sf)};
}
////////////////////////////////////////////////////////////////////////////////
// deduction guides
#if __cpp_deduction_guides >= 201703
deferred() -> deferred<ignoreSF>; deferred() -> deferred<ignoreSF>;
template <class SF> template <class SF>
deferred(SF) -> deferred<SF>; deferred(SF) -> deferred<SF>;
template <Sender<none_tag> Wrapped> PUSHMI_TEMPLATE(class Wrapped)
(requires Sender<Wrapped, is_none<>>)
deferred(Wrapped) -> deferred(Wrapped) ->
deferred<detail::erase_deferred_t, std::exception_ptr>; deferred<detail::erase_deferred_t, std::exception_ptr>;
template <Sender<none_tag> Data, class DSF> PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_none<>>)
deferred(Data, DSF) -> deferred<Data, DSF>; deferred(Data, DSF) -> deferred<Data, DSF>;
#endif
template <class E = std::exception_ptr> template <class E = std::exception_ptr>
using any_deferred = deferred<detail::erase_deferred_t, E>; using any_deferred = deferred<detail::erase_deferred_t, E>;
// template <SenderTo<any_none<std::exception_ptr>, none_tag> Wrapped> // template <SenderTo<any_none<std::exception_ptr>, is_none<>> Wrapped>
// auto erase_cast(Wrapped w) { // auto erase_cast(Wrapped w) {
// return deferred<detail::erase_deferred_t, std::exception_ptr>{std::move(w)}; // return deferred<detail::erase_deferred_t, std::exception_ptr>{std::move(w)};
// } // }
// //
// template <class E, SenderTo<any_none<E>, none_tag> Wrapped> // template <class E, SenderTo<any_none<E>, is_none<>> Wrapped>
// requires Same<none_tag, sender_category_t<Wrapped>> // requires Same<is_none<>, properties_t<Wrapped>>
// auto erase_cast(Wrapped w) { // auto erase_cast(Wrapped w) {
// return deferred<detail::erase_deferred_t, E>{std::move(w)}; // return deferred<detail::erase_deferred_t, E>{std::move(w)};
// } // }
......
This diff is collapsed.
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include <functional>
#include <meta/meta.hpp>
#include "../traits.h"
#include "../forwards.h"
namespace pushmi {
namespace detail {
struct placeholder;
template <class T, class Args, class = void>
struct substitute {
using type = T;
};
template <std::size_t I, class Args>
struct substitute<placeholder[I], Args>
: meta::lazy::let<
meta::defer<std::decay_t, meta::lazy::at<Args, meta::size_t<I-1>>>> {
};
template <std::size_t I, class Args>
struct substitute<placeholder(&&)[I], Args>
: meta::lazy::at<Args, meta::size_t<I-1>> {
};
template <template <class...> class R, class... Ts, class Args>
struct substitute<R<Ts...>, Args, meta::void_<R<meta::_t<substitute<Ts, Args>>...>>> {
using type = R<meta::_t<substitute<Ts, Args>>...>;
};
template <class Fn, class Requirements>
struct constrained_fn : Fn {
constrained_fn() = default;
constrained_fn(Fn fn) : Fn(std::move(fn)) {}
PUSHMI_TEMPLATE (class... Ts)
(requires Invocable<Fn&, Ts...> &&
(bool)meta::_t<substitute<Requirements, meta::list<Ts...>>>{})
decltype(auto) operator()(Ts&&... ts) noexcept(noexcept(std::declval<Fn&>()((Ts&&) ts...))) {
return static_cast<Fn&>(*this)((Ts&&) ts...);
}
PUSHMI_TEMPLATE (class... Ts)
(requires Invocable<const Fn&, Ts...> &&
(bool)meta::_t<substitute<Requirements, meta::list<Ts...>>>{})
decltype(auto) operator()(Ts&&... ts) const noexcept(noexcept(std::declval<const Fn&>()((Ts&&) ts...))) {
return static_cast<const Fn&>(*this)((Ts&&) ts...);
}
};
struct constrain_fn {
template <class Requirements, class Fn>
constexpr auto operator()(Requirements, Fn fn) const {
return constrained_fn<Fn, Requirements>{std::move(fn)};
}
};
} // namespace detail
using _1 = detail::placeholder[1];
using _2 = detail::placeholder[2];
using _3 = detail::placeholder[3];
PUSHMI_INLINE_VAR constexpr const detail::constrain_fn constrain {};
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
// Usage:
//
// PUSHMI_IF_CONSTEXPR((condition)(
// stmt1;
// stmt2;
// ) else (
// stmt3;
// stmt4;
// ))
//
// If the statements could potentially be ill-formed, you can give some
// part of the expression a dependent type by wrapping it in `id`. For
#define PUSHMI_COMMA ,
#define PUSHMI_EVAL(F, ...) F(__VA_ARGS__)
#define PUSHMI_STRIP(...) __VA_ARGS__
namespace pushmi {
namespace detail {
struct id_fn {
constexpr explicit operator bool() const noexcept {
return false;
}
template <class T>
constexpr T&& operator()(T&& t) const noexcept {
return (T&&) t;
}
};
}
}
#if __cpp_if_constexpr >= 201606
#define PUSHMI_IF_CONSTEXPR(LIST)\
if constexpr (::pushmi::detail::id_fn id = {}) {} \
else if constexpr PUSHMI_EVAL(PUSHMI_IF_CONSTEXPR_ELSE_, PUSHMI_IF_CONSTEXPR_IF_ LIST)
#define PUSHMI_IF_CONSTEXPR_RETURN(LIST)\
PUSHMI_IF_CONSTEXPR(LIST)\
/**/
#define PUSHMI_IF_CONSTEXPR_IF_(...) \
(__VA_ARGS__) PUSHMI_COMMA PUSHMI_IF_CONSTEXPR_THEN_
#define PUSHMI_IF_CONSTEXPR_THEN_(...) \
({__VA_ARGS__}) PUSHMI_COMMA
#define PUSHMI_IF_CONSTEXPR_ELSE_(A, B, C) \
A PUSHMI_STRIP B PUSHMI_IF_CONSTEXPR_ ## C
#define PUSHMI_IF_CONSTEXPR_else(...) \
else {__VA_ARGS__}
#else
#include <type_traits>
#define PUSHMI_IF_CONSTEXPR(LIST)\
PUSHMI_EVAL(PUSHMI_IF_CONSTEXPR_ELSE_, PUSHMI_IF_CONSTEXPR_IF_ LIST)\
/**/
#define PUSHMI_IF_CONSTEXPR_RETURN(LIST)\
return PUSHMI_EVAL(PUSHMI_IF_CONSTEXPR_ELSE_, PUSHMI_IF_CONSTEXPR_IF_ LIST)\
/**/
#define PUSHMI_IF_CONSTEXPR_IF_(...) \
(::pushmi::detail::select<bool(__VA_ARGS__)>() ->* PUSHMI_IF_CONSTEXPR_THEN_ \
/**/
#define PUSHMI_IF_CONSTEXPR_THEN_(...) \
([&](auto id)->decltype(auto){__VA_ARGS__})) PUSHMI_COMMA \
/**/
#define PUSHMI_IF_CONSTEXPR_ELSE_(A, B) \
A ->* PUSHMI_IF_CONSTEXPR_ ## B \
/**/
#define PUSHMI_IF_CONSTEXPR_else(...) ([&](auto id)->decltype(auto){__VA_ARGS__});
namespace pushmi {
namespace detail {
template <bool>
struct select {
template <class R, class = std::enable_if_t<!std::is_void<R>::value>>
struct eat_return {
R value_;
template <class T>
constexpr R operator->*(T&&) {
return static_cast<R&&>(value_);
}
};
struct eat {
template <class T>
constexpr void operator->*(T&&) {}
};
template <class T>
constexpr auto operator->*(T&& t) -> eat_return<decltype(t(::pushmi::detail::id_fn{}))> {
return {t(::pushmi::detail::id_fn{})};
}
template <class T>
constexpr auto operator->*(T&& t) const -> eat {
return t(::pushmi::detail::id_fn{}), void(), eat{};
}
};
template <>
struct select<false> {
struct eat {
template <class T>
constexpr auto operator->*(T&& t) -> decltype(auto) {
return t(::pushmi::detail::id_fn{});
}
};
template <class T>
constexpr eat operator->*(T&& t) {
return {};
}
};
}
}
#endif
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#if __cpp_lib_optional >= 201606
#include <optional>
#endif
#include <type_traits>
namespace pushmi {
namespace detail {
#if __cpp_lib_optional >= 201606
template <class T>
struct opt : private std::optional<T> {
opt() = default;
opt& operator=(T&& t) {
this->std::optional<T>::operator=(std::move(t));
return *this;
}
using std::optional<T>::operator*;
using std::optional<T>::operator bool;
};
#else
template <class T>
struct opt {
private:
bool empty_ = true;
std::aligned_union_t<0, T> data_;
T* ptr() {
return static_cast<T*>((void*)&data_);
}
const T* ptr() const {
return static_cast<const T*>((const void*)&data_);
}
void reset() {
if (!empty_) {
ptr()->~T();
empty_ = true;
}
}
public:
opt() = default;
opt(T&& t) noexcept(std::is_nothrow_move_constructible<T>::value) {
::new(ptr()) T(std::move(t));
empty_ = false;
}
opt(const T& t) {
::new(ptr()) T(t);
empty_ = false;
}
opt(opt&& that) noexcept(std::is_nothrow_move_constructible<T>::value) {
if (that) {
::new(ptr()) T(std::move(*that));
empty_ = false;
that.reset();
}
}
opt(const opt& that) {
if (that) {
::new(ptr()) T(*that);
empty_ = false;
}
}
~opt() { reset(); }
opt& operator=(opt&& that)
noexcept(std::is_nothrow_move_constructible<T>::value &&
std::is_nothrow_move_assignable<T>::value) {
if (*this && that) {
**this = std::move(*that);
that.reset();
} else if (*this) {
reset();
} else if (that) {
::new(ptr()) T(std::move(*that));
empty_ = false;
}
return *this;
}
opt& operator=(const opt& that) {
if (*this && that) {
**this = *that;
} else if (*this) {
reset();
} else if (that) {
::new(ptr()) T(*that);
empty_ = false;
}
return *this;
}
opt& operator=(T&& t) noexcept(std::is_nothrow_move_constructible<T>::value &&
std::is_nothrow_move_assignable<T>::value) {
if (*this)
**this = std::move(t);
else {
::new(ptr()) T(std::move(t));
empty_ = false;
}
return *this;
}
opt& operator=(const T& t) {
if (*this)
**this = t;
else {
::new(ptr()) T(t);
empty_ = false;
}
return *this;
}
explicit operator bool() const noexcept {
return !empty_;
}
T& operator*() noexcept {
return *ptr();
}
const T& operator*() const noexcept {
return *ptr();
}
};
#endif
} // namespace detail
} // namespace pushmi
...@@ -11,72 +11,52 @@ ...@@ -11,72 +11,52 @@
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
template<class E, class TP>
template<class TP> struct any_time_executor_ref_base {
struct any_time_executor_ref_vtable {
TP (*now_)(void*);
void (*submit_)(void*, TP, void*);
};
template <class E, class TP, class Other, class Wrapped>
auto any_time_executor_ref_vtable_v() {
static constexpr any_time_executor_ref_vtable<TP> const vtable_v {
+[](void* pobj) { return ::pushmi::now(*static_cast<Wrapped*>(pobj)); },
+[](void* pobj, TP tp, void* s) {
return ::pushmi::submit(
*static_cast<Wrapped*>(pobj),
tp,
std::move(*static_cast<single<Other, E>*>(s)));
}
};
return &vtable_v;
};
} // namespace detail
template<class E, class TP, int i>
struct any_time_executor_ref {
private: private:
// use two instances to resolve recurive type definition. friend any_time_executor_ref<E, TP, 0>;
using This = any_time_executor_ref<E, TP, i>; friend any_time_executor_ref<E, TP, 1>;
using Other = any_time_executor_ref<E, TP, i == 0 ? 1 : 0>; using Other = any_time_executor_ref<E, TP, 1>;
void* pobj_;
struct vtable {
TP (*now_)(void*);
void (*submit_)(void*, TP, void*);
} const *vptr_;
template <class T, class U = std::decay_t<T>> template <class T, class U = std::decay_t<T>>
using wrapped_t = using wrapped_t =
std::enable_if_t< std::enable_if_t<!std::is_base_of<any_time_executor_ref_base, U>::value, U>;
!std::is_same_v<U, This> &&
!std::is_same_v<U, Other>, U>;
void* pobj_;
detail::any_time_executor_ref_vtable<TP> const *vptr_;
public: public:
using sender_category = single_tag; using properties = property_set<is_time<>, is_single<>>;
any_time_executor_ref() = delete; any_time_executor_ref_base() = delete;
any_time_executor_ref_base(const any_time_executor_ref_base&) = default;
template<int n>
any_time_executor_ref(any_time_executor_ref<E, TP, n>&& o) : PUSHMI_TEMPLATE (class Wrapped)
pobj_(o.pobj_), vptr_(o.vptr_) { (requires TimeSender<wrapped_t<Wrapped>, is_single<>>)
o.pobj_ = nullptr; // (requires TimeSenderTo<wrapped_t<Wrapped>, single<Other, E>>)
o.vptr_ = nullptr; any_time_executor_ref_base(Wrapped& w) {
};
template<int n>
any_time_executor_ref(const any_time_executor_ref<E, TP, n>& o) :
pobj_(o.pobj_), vptr_(o.vptr_) {};
template <class Wrapped, TimeSender<single_tag> W = wrapped_t<Wrapped>>
// requires TimeSenderTo<W, single<Other, E>>
any_time_executor_ref(Wrapped& w) {
// This can't be a requirement because it asks if submit(w, now(w), single<T,E>) // This can't be a requirement because it asks if submit(w, now(w), single<T,E>)
// is well-formed (where T is an alias for any_time_executor_ref). If w // is well-formed (where T is an alias for any_time_executor_ref). If w
// has a submit that is constrained with SingleReceiver<single<T, E>, T'&, E'>, that // has a submit that is constrained with SingleReceiver<single<T, E>, T'&, E'>, that
// will ask whether value(single<T,E>, T'&) is well-formed. And *that* will // will ask whether value(single<T,E>, T'&) is well-formed. And *that* will
// ask whether T'& is convertible to T. That brings us right back to this // ask whether T'& is convertible to T. That brings us right back to this
// constructor. Constraint recursion! // constructor. Constraint recursion!
static_assert(TimeSenderTo<W, single<Other, E>>); static_assert(TimeSenderTo<Wrapped, single<Other, E>>);
if constexpr((bool)TimeSenderTo<W, single<Other, E>>) { struct s {
pobj_ = std::addressof(w); static TP now(void* pobj) {
vptr_ = detail::any_time_executor_ref_vtable_v<E, TP, Other, Wrapped>(); return ::pushmi::now(*static_cast<Wrapped*>(pobj));
} }
static void submit(void* pobj, TP tp, void* s) {
return ::pushmi::submit(
*static_cast<Wrapped*>(pobj),
tp,
std::move(*static_cast<single<Other, E>*>(s)));
}
};
static const vtable vtbl{s::now, s::submit};
pobj_ = std::addressof(w);
vptr_ = &vtbl;
} }
std::chrono::system_clock::time_point now() { std::chrono::system_clock::time_point now() {
return vptr_->now_(pobj_); return vptr_->now_(pobj_);
...@@ -90,14 +70,45 @@ public: ...@@ -90,14 +70,45 @@ public:
vptr_->submit_(pobj_, tp, &s); vptr_->submit_(pobj_, tp, &s);
} }
}; };
} // namespace detail
using archtype_any_time_executor_ref = any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point, 0>; template<class E, class TP, int i>
struct any_time_executor_ref : detail::any_time_executor_ref_base<E, TP> {
template <class E = std::exception_ptr, class TP = std::chrono::system_clock::time_point> using detail::any_time_executor_ref_base<E, TP>::any_time_executor_ref_base;
any_time_executor_ref()->any_time_executor_ref<E, TP, 0>; any_time_executor_ref(const detail::any_time_executor_ref_base<E, TP>& o)
: detail::any_time_executor_ref_base<E, TP>(o) {}
};
template <class Wrapped, class E = std::exception_ptr, class TP = std::chrono::system_clock::time_point> ////////////////////////////////////////////////////////////////////////////////
any_time_executor_ref(Wrapped)->any_time_executor_ref<E, TP, 0>; // make_any_time_executor_ref
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
auto make_any_time_executor_ref() -> any_time_executor_ref<E, TP> {
return any_time_executor_ref<E, TP, 0>{};
}
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point,
class Wrapped>
auto make_any_time_executor_ref(Wrapped w) -> any_time_executor_ref<E, TP> {
return any_time_executor_ref<E, TP, 0>{std::move(w)};
}
////////////////////////////////////////////////////////////////////////////////
// deduction guides
#if __cpp_deduction_guides >= 201703
any_time_executor_ref() ->
any_time_executor_ref<
std::exception_ptr,
std::chrono::system_clock::time_point>;
template <class Wrapped>
any_time_executor_ref(Wrapped) ->
any_time_executor_ref<
std::exception_ptr,
std::chrono::system_clock::time_point>;
#endif
template<class E, class TP> template<class E, class TP>
struct any_time_executor : struct any_time_executor :
...@@ -107,10 +118,36 @@ struct any_time_executor : ...@@ -107,10 +118,36 @@ struct any_time_executor :
any_time_single_deferred; any_time_single_deferred;
}; };
template <class E = std::exception_ptr, class TP = std::chrono::system_clock::time_point> ////////////////////////////////////////////////////////////////////////////////
any_time_executor()->any_time_executor<E, TP>; // make_any_time_executor
template <
template <class E = std::exception_ptr, class TP = std::chrono::system_clock::time_point, class Wrapped> class E = std::exception_ptr,
any_time_executor(Wrapped)->any_time_executor<E, TP>; class TP = std::chrono::system_clock::time_point>
auto make_any_time_executor() -> any_time_executor<E, TP> {
return any_time_executor<E, TP>{};
}
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point,
class Wrapped>
auto make_any_time_executor(Wrapped w) -> any_time_executor<E, TP> {
return any_time_executor<E, TP>{std::move(w)};
}
////////////////////////////////////////////////////////////////////////////////
// deduction guides
#if __cpp_deduction_guides >= 201703
any_time_executor() ->
any_time_executor<
std::exception_ptr,
std::chrono::system_clock::time_point>;
template <class Wrapped>
any_time_executor(Wrapped) ->
any_time_executor<
std::exception_ptr,
std::chrono::system_clock::time_point>;
#endif
} // namespace pushmi } // namespace pushmi
...@@ -17,60 +17,67 @@ class flow_single_deferred<V, PE, E> { ...@@ -17,60 +17,67 @@ class flow_single_deferred<V, PE, E> {
template <class Wrapped> template <class Wrapped>
static constexpr bool insitu() { static constexpr bool insitu() {
return sizeof(Wrapped) <= sizeof(data::buffer_) && return sizeof(Wrapped) <= sizeof(data::buffer_) &&
std::is_nothrow_move_constructible_v<Wrapped>; std::is_nothrow_move_constructible<Wrapped>::value;
} }
enum struct op { destroy, move };
struct vtable { struct vtable {
void (*op_)(op, data&, data*) = +[](op, data&, data*) {}; static void s_op(data&, data*) {}
void (*submit_)(data&, flow_single<V, PE, E>) = static void s_submit(data&, flow_single<V, PE, E>) {}
+[](data&, flow_single<V, PE, E>) {}; void (*op_)(data&, data*) = s_op;
static constexpr vtable const noop_ = {}; void (*submit_)(data&, flow_single<V, PE, E>) = s_submit;
static constexpr vtable const noop_ {};
} const* vptr_ = &vtable::noop_; } const* vptr_ = &vtable::noop_;
template <class Wrapped, bool = insitu<Wrapped>()>
static constexpr vtable const vtable_v = {
+[](op o, data& src, data* dst) {
switch (o) {
case op::move:
dst->pobj_ = std::exchange(src.pobj_, nullptr);
case op::destroy:
delete static_cast<Wrapped const*>(src.pobj_);
}
},
+[](data& src, flow_single<V, PE, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
template <class T, class U = std::decay_t<T>> template <class T, class U = std::decay_t<T>>
using wrapped_t = using wrapped_t =
std::enable_if_t<!std::is_same_v<U, flow_single_deferred>, U>; std::enable_if_t<!std::is_same<U, flow_single_deferred>::value, U>;
public: public:
using sender_category = flow_tag; using properties = property_set<is_sender<>, is_flow<>, is_single<>>;
flow_single_deferred() = default; flow_single_deferred() = default;
flow_single_deferred(flow_single_deferred&& that) noexcept flow_single_deferred(flow_single_deferred&& that) noexcept
: flow_single_deferred() { : flow_single_deferred() {
that.vptr_->op_(op::move, that.data_, &data_); that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_); std::swap(that.vptr_, vptr_);
} }
template <class Wrapped> PUSHMI_TEMPLATE (class Wrapped)
requires FlowSingleSender< (requires FlowSender<wrapped_t<Wrapped>, is_single<>>
wrapped_t<Wrapped>, PUSHMI_BROKEN_SUBSUMPTION(&& !insitu<Wrapped>()))
flow_single<V, PE, E>, explicit flow_single_deferred(Wrapped obj) : flow_single_deferred() {
any_none<PE>, struct s {
V, static void op(data& src, data* dst) {
E> if (dst)
explicit flow_single_deferred(Wrapped obj) dst->pobj_ = std::exchange(src.pobj_, nullptr);
: flow_single_deferred() { delete static_cast<Wrapped const*>(src.pobj_);
if constexpr (insitu<Wrapped>()) }
new (data_.buffer_) Wrapped(std::move(obj)); static void submit(data& src, flow_single<V, PE, E> out) {
else ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
data_.pobj_ = new Wrapped(std::move(obj)); }
vptr_ = &vtable_v<Wrapped>; };
static const vtable vtbl{s::op, s::submit};
data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl;
}
PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>, is_single<>> && insitu<Wrapped>())
explicit flow_single_deferred(Wrapped obj) noexcept : flow_single_deferred() {
struct s {
static void op(data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
static void submit(data& src, flow_single<V, PE, E> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_),
std::move(out));
}
};
static const vtable vtbl{s::op, s::submit};
new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl;
} }
~flow_single_deferred() { ~flow_single_deferred() {
vptr_->op_(op::destroy, data_, nullptr); vptr_->op_(data_, nullptr);
} }
flow_single_deferred& operator=(flow_single_deferred&& that) noexcept { flow_single_deferred& operator=(flow_single_deferred&& that) noexcept {
this->~flow_single_deferred(); this->~flow_single_deferred();
...@@ -86,51 +93,43 @@ class flow_single_deferred<V, PE, E> { ...@@ -86,51 +93,43 @@ class flow_single_deferred<V, PE, E> {
template <class V, class PE, class E> template <class V, class PE, class E>
constexpr typename flow_single_deferred<V, PE, E>::vtable const constexpr typename flow_single_deferred<V, PE, E>::vtable const
flow_single_deferred<V, PE, E>::vtable::noop_; flow_single_deferred<V, PE, E>::vtable::noop_;
template <class V, class PE, class E>
template <class Wrapped, bool Big>
constexpr typename flow_single_deferred<V, PE, E>::vtable const
flow_single_deferred<V, PE, E>::vtable_v;
template <class V, class PE, class E>
template <class Wrapped>
constexpr typename flow_single_deferred<V, PE, E>::vtable const
flow_single_deferred<V, PE, E>::vtable_v<Wrapped, true> = {
+[](op o, data& src, data* dst) {
switch (o) {
case op::move:
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
case op::destroy:
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
},
+[](data& src, flow_single<V, PE, E> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
}
};
template <class SF> template <class SF>
class flow_single_deferred<SF> { class flow_single_deferred<SF> {
SF sf_; SF sf_;
public: public:
using sender_category = flow_tag; using properties = property_set<is_sender<>, is_flow<>, is_single<>>;
constexpr flow_single_deferred() = default; constexpr flow_single_deferred() = default;
constexpr explicit flow_single_deferred(SF sf) constexpr explicit flow_single_deferred(SF sf)
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
template <Receiver<flow_tag> Out> PUSHMI_TEMPLATE(class Out)
requires Invocable<SF&, Out> (requires Receiver<Out, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
sf_(std::move(out)); sf_(std::move(out));
} }
}; };
////////////////////////////////////////////////////////////////////////////////
// make_flow_single_deferred
inline auto make_flow_single_deferred() -> flow_single_deferred<ignoreSF> {
return flow_single_deferred<ignoreSF>{};
}
template <class SF>
auto make_flow_single_deferred(SF sf) -> flow_single_deferred<SF> {
return flow_single_deferred<SF>(std::move(sf));
}
////////////////////////////////////////////////////////////////////////////////
// deduction guides
#if __cpp_deduction_guides >= 201703
flow_single_deferred() -> flow_single_deferred<ignoreSF>; flow_single_deferred() -> flow_single_deferred<ignoreSF>;
template <class SF> template <class SF>
flow_single_deferred(SF) -> flow_single_deferred<SF>; flow_single_deferred(SF) -> flow_single_deferred<SF>;
#endif
template <class V, class PE = std::exception_ptr, class E = PE> template <class V, class PE = std::exception_ptr, class E = PE>
using any_flow_single_deferred = flow_single_deferred<V, PE, E>; using any_flow_single_deferred = flow_single_deferred<V, PE, E>;
......
...@@ -4,50 +4,81 @@ ...@@ -4,50 +4,81 @@
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include <exception>
#include <chrono>
#include "traits.h" #include "traits.h"
namespace pushmi { namespace pushmi {
// tag types // property_set
struct silent_tag;
struct none_tag;
struct single_tag;
struct flow_tag;
template <class> template <class T, class = void>
struct sender_traits; struct property_traits;
template <class> template <class T>
struct receiver_traits; struct property_set_traits;
template <SemiMovable... TN> template<class... PropertyN>
struct property_set;
// trait & tag types
template<class...TN>
struct is_silent;
template<class...TN>
struct is_none;
template<class...TN>
struct is_single;
template<class...TN>
struct is_many;
template<class...TN>
struct is_flow;
template<class...TN>
struct is_receiver;
template<class...TN>
struct is_sender;
template<class...TN>
struct is_time;
template<class...TN>
struct is_constrained;
// implementation types
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class none; class none;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class deferred; class deferred;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class single; class single;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class single_deferred; class single_deferred;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class time_single_deferred; class time_single_deferred;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_single; class flow_single;
template <SemiMovable... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_single_deferred; class flow_single_deferred;
template<class E, class TP, int i = 0> template<
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point,
int i = 0>
struct any_time_executor_ref; struct any_time_executor_ref;
namespace operators {} namespace operators {}
namespace extension_operators {} namespace extension_operators {}
namespace aliases { namespace aliases {
namespace v = ::pushmi; namespace v = ::pushmi;
namespace mi = ::pushmi;
namespace op = ::pushmi::operators; namespace op = ::pushmi::operators;
namespace ep = ::pushmi::extension_operators; namespace ep = ::pushmi::extension_operators;
} }
......
...@@ -13,22 +13,20 @@ namespace pushmi { ...@@ -13,22 +13,20 @@ namespace pushmi {
// //
struct __new_thread_submit { struct __new_thread_submit {
template<Regular TP, Receiver Out> PUSHMI_TEMPLATE(class TP, class Out)
void operator()(TP at, Out out) const; (requires Regular<TP> && Receiver<Out>)
void operator()(TP at, Out out) const {
std::thread t{[at = std::move(at), out = std::move(out)]() mutable {
auto tr = trampoline();
::pushmi::submit(tr, std::move(at), std::move(out));
}};
// pass ownership of thread to out
t.detach();
}
}; };
template<Regular TP, Receiver Out>
void __new_thread_submit::operator()(TP at, Out out) const {
std::thread t{[at = std::move(at), out = std::move(out)]() mutable {
auto tr = trampoline();
::pushmi::submit(tr, std::move(at), std::move(out));
}};
// pass ownership of thread to out
t.detach();
}
inline auto new_thread() { inline auto new_thread() {
return time_single_deferred{__new_thread_submit{}}; return make_time_single_deferred(__new_thread_submit{});
} }
} }
...@@ -6,8 +6,11 @@ ...@@ -6,8 +6,11 @@
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include <meta/meta.hpp>
#include "../deferred.h" #include "../deferred.h"
#include "../single_deferred.h" #include "../single_deferred.h"
#include "../detail/functional.h"
namespace pushmi { namespace pushmi {
...@@ -15,20 +18,20 @@ namespace operators { ...@@ -15,20 +18,20 @@ namespace operators {
template <class V> template <class V>
auto empty() { auto empty() {
return single_deferred{ return make_single_deferred(
[]<class Out>(Out out) mutable PUSHMI_VOID_LAMBDA_REQUIRES( constrain(lazy::SingleReceiver<_1, V>, [](auto out) mutable {
SingleReceiver<Out, V>){ ::pushmi::set_done(out);
::pushmi::set_done(out); })
}}; );
} }
inline auto empty() { inline auto empty() {
return deferred{[]<class Out>(Out out) mutable PUSHMI_VOID_LAMBDA_REQUIRES( return make_deferred(
NoneReceiver<Out>){ constrain(lazy::NoneReceiver<_1>, [](auto out) mutable {
::pushmi::set_done(out); ::pushmi::set_done(out);
}}; })
);
} }
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
...@@ -6,18 +6,22 @@ ...@@ -6,18 +6,22 @@
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "../single.h" #include "../single_deferred.h"
namespace pushmi { namespace pushmi {
namespace operators { namespace operators {
template <class V> PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
auto just(V v) { auto just(V v) {
return single_deferred{[v = std::move(v)]<class Out>( return make_single_deferred(
Out out) mutable PUSHMI_VOID_LAMBDA_REQUIRES(SingleReceiver<Out, V>){ constrain(lazy::SingleReceiver<_1, V>,
::pushmi::set_value(out, std::move(v)); [v = std::move(v)](auto out) mutable {
}}; ::pushmi::set_value(out, std::move(v));
}
)
);
} }
} // namespace operators } // namespace operators
......
...@@ -10,41 +10,44 @@ ...@@ -10,41 +10,44 @@
namespace pushmi { namespace pushmi {
namespace operators {
namespace detail { namespace detail {
struct on_fn { struct on_fn {
template <class ExecutorFactory> PUSHMI_TEMPLATE(class ExecutorFactory)
auto operator()(ExecutorFactory ef) const; (requires Invocable<ExecutorFactory&>)
auto operator()(ExecutorFactory ef) const {
return constrain(lazy::Sender<_1>, [ef = std::move(ef)](auto in) {
using In = decltype(in);
return ::pushmi::detail::deferred_from<In, single<>>(
std::move(in),
::pushmi::detail::submit_transform_out<In>(
constrain(lazy::SenderTo<In, _2>, [ef](In& in, auto out) {
auto exec = ef();
::pushmi::submit(exec, ::pushmi::now(exec),
::pushmi::make_single([in = in, out = std::move(out)](auto) mutable {
::pushmi::submit(in, std::move(out));
})
);
}),
constrain(lazy::TimeSenderTo<In, _3>, [ef](In& in, auto at, auto out) {
auto exec = ef();
::pushmi::submit(exec, at,
::pushmi::on_value([in = in, at, out = std::move(out)](auto) mutable {
::pushmi::submit(in, at, std::move(out));
})
);
})
)
);
});
}
}; };
template <class ExecutorFactory>
auto on_fn::operator()(ExecutorFactory ef) const {
return [ef = std::move(ef)]<class In>(In in) {
return ::pushmi::detail::deferred_from<In, single<>>(
std::move(in),
::pushmi::detail::submit_transform_out<In>(
[ef]<class Out>(In& in, Out out) {
auto exec = ef();
::pushmi::submit(exec, ::pushmi::now(exec), ::pushmi::single{[in = in, out = std::move(out)](auto) mutable {
::pushmi::submit(in, std::move(out));
}});
},
[ef]<class TP, class Out>(In& in, TP at, Out out) {
auto exec = ef();
::pushmi::submit(exec, at, ::pushmi::on_value{[in = in, at, out = std::move(out)](auto) mutable {
::pushmi::submit(in, at, std::move(out));
}});
}
)
);
};
}
} // namespace detail } // namespace detail
inline constexpr detail::on_fn on{}; namespace operators {
PUSHMI_INLINE_VAR constexpr detail::on_fn on{};
} // namespace operators } // namespace operators
......
...@@ -12,8 +12,6 @@ ...@@ -12,8 +12,6 @@
namespace pushmi { namespace pushmi {
namespace operators {
namespace detail { namespace detail {
struct transform_fn { struct transform_fn {
...@@ -23,18 +21,21 @@ struct transform_fn { ...@@ -23,18 +21,21 @@ struct transform_fn {
template <class... FN> template <class... FN>
auto transform_fn::operator()(FN... fn) const { auto transform_fn::operator()(FN... fn) const {
auto f = overload{std::move(fn)...}; auto f = overload(std::move(fn)...);
return [f = std::move(f)]<class In>(In in) { return constrain(lazy::Sender<_1>, [f = std::move(f)](auto in) {
using In = decltype(in);
// copy 'f' to allow multiple calls to connect to multiple 'in' // copy 'f' to allow multiple calls to connect to multiple 'in'
return ::pushmi::detail::deferred_from<In, single<>>( return ::pushmi::detail::deferred_from<In, single<>>(
std::move(in), std::move(in),
::pushmi::detail::submit_transform_out<In>( ::pushmi::detail::submit_transform_out<In>(
[f]<class Out>(Out out) { constrain(lazy::Receiver<_1>, [f](auto out) {
using Out = decltype(out);
return ::pushmi::detail::out_from_fn<In>()( return ::pushmi::detail::out_from_fn<In>()(
std::move(out), std::move(out),
// copy 'f' to allow multiple calls to submit // copy 'f' to allow multiple calls to submit
on_value{ on_value(
[f]<class V>(Out& out, V&& v) { [f](Out& out, auto&& v) {
using V = decltype(v);
using Result = decltype(f((V&&) v)); using Result = decltype(f((V&&) v));
static_assert(SemiMovable<Result>, static_assert(SemiMovable<Result>,
"none of the functions supplied to transform can convert this value"); "none of the functions supplied to transform can convert this value");
...@@ -42,18 +43,18 @@ auto transform_fn::operator()(FN... fn) const { ...@@ -42,18 +43,18 @@ auto transform_fn::operator()(FN... fn) const {
"Result of value transform cannot be delivered to Out"); "Result of value transform cannot be delivered to Out");
::pushmi::set_value(out, f((V&&) v)); ::pushmi::set_value(out, f((V&&) v));
} }
} )
); );
} })
) )
); );
}; });
} }
} // namespace detail } // namespace detail
inline constexpr detail::transform_fn transform{}; namespace operators {
PUSHMI_INLINE_VAR constexpr detail::transform_fn transform{};
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
...@@ -6,16 +6,30 @@ ...@@ -6,16 +6,30 @@
#include "traits.h" #include "traits.h"
template <class In, pushmi::Invocable<In> Op> PUSHMI_TEMPLATE (class In, class Op)
(requires pushmi::Invocable<Op&, In>)
decltype(auto) operator|(In&& in, Op op) { decltype(auto) operator|(In&& in, Op op) {
return op((In&&) in); return op((In&&) in);
} }
namespace pushmi { namespace pushmi {
template<class T, class... FN> PUSHMI_INLINE_VAR constexpr struct pipe_fn {
auto pipe(T t, FN... fn) -> decltype((t | ... | fn)) { #if __cpp_fold_expressions >= 201603
return (t | ... | fn); template<class T, class... FN>
} auto operator()(T t, FN... fn) const -> decltype((t | ... | fn)) {
return (t | ... | fn);
}
#else
template<class T, class F>
auto operator()(T t, F f) const -> decltype(t | f) {
return t | f;
}
template<class T, class F, class... FN, class This = pipe_fn>
auto operator()(T t, F f, FN... fn) const -> decltype(This()((t | f), fn...)) {
return This()((t | f), fn...);
}
#endif
} const pipe {};
} // namespace pushmi } // namespace pushmi
This diff is collapsed.
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