Commit ebd46723 authored by Kirk Shoop's avatar Kirk Shoop Committed by Facebook Github Bot

adding executor() method (#50)

fbshipit-source-id: 73368129775f17ebed8f74cb39688b2b6a9d73e4
parent 82c0f564
...@@ -6,6 +6,7 @@ project(pushmi-project CXX) ...@@ -6,6 +6,7 @@ project(pushmi-project CXX)
option(PUSHMI_USE_CONCEPTS_EMULATION "Use C++14 Concepts Emulation" ON) option(PUSHMI_USE_CONCEPTS_EMULATION "Use C++14 Concepts Emulation" ON)
option(PUSHMI_USE_CPP_2A "Use C++2a with concepts emulation" OFF) option(PUSHMI_USE_CPP_2A "Use C++2a with concepts emulation" OFF)
option(PUSHMI_USE_CPP_17 "Use C++17 with concepts emulation" OFF) option(PUSHMI_USE_CPP_17 "Use C++17 with concepts emulation" OFF)
option(PUSHMI_ONE_TEST_BINARY "Compile all the tests into one binary" OFF)
FIND_PACKAGE (Threads REQUIRED) FIND_PACKAGE (Threads REQUIRED)
......
...@@ -79,18 +79,11 @@ struct countdownnone { ...@@ -79,18 +79,11 @@ struct countdownnone {
void operator()() const; void operator()() const;
}; };
struct inline_executor {
using properties = mi::property_set<mi::is_sender<>, mi::is_single<>>;
template<class Out>
void submit(Out out) {
::mi::set_value(out, *this);
}
};
struct inline_time_executor { struct inline_time_executor {
using properties = mi::property_set<mi::is_time<>, mi::is_single<>>; using properties = mi::property_set<mi::is_time<>, mi::is_executor<>, mi::is_single<>>;
std::chrono::system_clock::time_point now() { return std::chrono::system_clock::now(); } std::chrono::system_clock::time_point now() { return std::chrono::system_clock::now(); }
auto executor() { return *this; }
template<class Out> template<class Out>
void submit(std::chrono::system_clock::time_point at, Out out) { void submit(std::chrono::system_clock::time_point at, Out out) {
std::this_thread::sleep_until(at); std::this_thread::sleep_until(at);
...@@ -98,8 +91,18 @@ struct inline_time_executor { ...@@ -98,8 +91,18 @@ struct inline_time_executor {
} }
}; };
struct inline_executor {
using properties = mi::property_set<mi::is_sender<>, mi::is_single<>>;
auto executor() { return inline_time_executor{}; }
template<class Out>
void submit(Out out) {
::mi::set_value(out, *this);
}
};
struct inline_executor_none { struct inline_executor_none {
using properties = mi::property_set<mi::is_sender<>, mi::is_none<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_none<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
::mi::set_done(out); ::mi::set_done(out);
...@@ -117,6 +120,7 @@ struct inline_executor_flow_single { ...@@ -117,6 +120,7 @@ struct inline_executor_flow_single {
CancellationFactory cf; CancellationFactory cf;
using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
...@@ -181,6 +185,7 @@ using inline_executor_flow_single_entangled = inline_executor_flow_single<entang ...@@ -181,6 +185,7 @@ using inline_executor_flow_single_entangled = inline_executor_flow_single<entang
struct inline_executor_flow_single_ignore { struct inline_executor_flow_single_ignore {
using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
// pass reference for cancellation. // pass reference for cancellation.
...@@ -200,6 +205,7 @@ struct inline_executor_flow_many { ...@@ -200,6 +205,7 @@ struct inline_executor_flow_many {
using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_many<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_many<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
...@@ -244,6 +250,7 @@ struct inline_executor_flow_many { ...@@ -244,6 +250,7 @@ struct inline_executor_flow_many {
struct inline_executor_flow_many_ignore { struct inline_executor_flow_many_ignore {
using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_many<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_many<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
// pass reference for cancellation. // pass reference for cancellation.
...@@ -257,6 +264,7 @@ struct inline_executor_flow_many_ignore { ...@@ -257,6 +264,7 @@ struct inline_executor_flow_many_ignore {
struct inline_executor_many { struct inline_executor_many {
using properties = mi::property_set<mi::is_sender<>, mi::is_many<>>; using properties = mi::property_set<mi::is_sender<>, mi::is_many<>>;
auto executor() { return inline_time_executor{}; }
template<class Out> template<class Out>
void submit(Out out) { void submit(Out out) {
::mi::set_next(out, *this); ::mi::set_next(out, *this);
......
...@@ -34,19 +34,20 @@ set(header_files ...@@ -34,19 +34,20 @@ set(header_files
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/boosters.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/boosters.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/piping.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/piping.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/none.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/none.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/executor.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many_sender.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/executor.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/inline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h"
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include <pushmi/executor.h> #include <pushmi/executor.h>
#include <pushmi/trampoline.h> #include <pushmi/trampoline.h>
#include <pushmi/time_single_sender.h>
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_ #define MAKE(x) x MAKE_
...@@ -23,14 +24,20 @@ using std::experimental::static_thread_pool; ...@@ -23,14 +24,20 @@ using std::experimental::static_thread_pool;
namespace execution = std::experimental::execution; namespace execution = std::experimental::execution;
template<class Executor> template<class Executor>
struct __pool_submit { struct pool_time_executor {
using properties = property_set<is_time<>, is_executor<>, is_single<>>;
using e_t = Executor; using e_t = Executor;
e_t e; e_t e;
explicit __pool_submit(e_t e) : e(std::move(e)) {} explicit pool_time_executor(e_t e) : e(std::move(e)) {}
auto now() {
return std::chrono::system_clock::now();
}
auto executor() { return *this; }
PUSHMI_TEMPLATE(class TP, class Out) PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out>) (requires Regular<TP> && Receiver<Out>)
void operator()(TP at, Out out) const { void submit(TP at, Out out) const {
e.execute([e = make_time_single_sender(*this), at = std::move(at), out = std::move(out)]() mutable { e.execute([e = *this, at = std::move(at), out = std::move(out)]() mutable {
std::this_thread::sleep_until(at); std::this_thread::sleep_until(at);
::pushmi::set_value(out, e); ::pushmi::set_value(out, e);
}); });
...@@ -45,7 +52,7 @@ public: ...@@ -45,7 +52,7 @@ public:
inline auto executor() { inline auto executor() {
auto exec = execution::require(p.executor(), execution::never_blocking, execution::oneway); auto exec = execution::require(p.executor(), execution::never_blocking, execution::oneway);
return MAKE(time_single_sender)(__pool_submit<decltype(exec)>{exec}); return pool_time_executor<decltype(exec)>{exec};
} }
inline void stop() {p.stop();} inline void stop() {p.stop();}
......
This diff is collapsed.
...@@ -87,6 +87,12 @@ struct ignoreStrtF { ...@@ -87,6 +87,12 @@ struct ignoreStrtF {
void operator()(detail::any) {} void operator()(detail::any) {}
}; };
struct trampolineEXF;
// see trampoline.h
// struct trampolineEXF {
// auto operator()() { return trampoline(); }
// };
struct ignoreSF { struct ignoreSF {
void operator()(detail::any) {} void operator()(detail::any) {}
...@@ -143,6 +149,13 @@ struct passDStrtF { ...@@ -143,6 +149,13 @@ struct passDStrtF {
} }
}; };
struct passDEXF {
PUSHMI_TEMPLATE(class Data)
(requires Sender<Data>)
auto operator()(Data& in) const noexcept {
return ::pushmi::executor(in);
}
};
struct passDSF { struct passDSF {
template <class Data, class Out> template <class Data, class Out>
...@@ -287,6 +300,17 @@ auto on_starting(Fns... fns) -> on_starting_fn<Fns...> { ...@@ -287,6 +300,17 @@ auto on_starting(Fns... fns) -> on_starting_fn<Fns...> {
return on_starting_fn<Fns...>{std::move(fns)...}; return on_starting_fn<Fns...>{std::move(fns)...};
} }
template <class Fn>
struct on_executor_fn : overload_fn<Fn> {
constexpr on_executor_fn() = default;
using overload_fn<Fn>::overload_fn;
};
template <class Fn>
auto on_executor(Fn fn) -> on_executor_fn<Fn> {
return on_executor_fn<Fn>{std::move(fn)};
}
template <class... Fns> template <class... Fns>
struct on_submit_fn : overload_fn<Fns...> { struct on_submit_fn : overload_fn<Fns...> {
constexpr on_submit_fn() = default; constexpr on_submit_fn() = default;
...@@ -298,4 +322,15 @@ auto on_submit(Fns... fns) -> on_submit_fn<Fns...> { ...@@ -298,4 +322,15 @@ auto on_submit(Fns... fns) -> on_submit_fn<Fns...> {
return on_submit_fn<Fns...>{std::move(fns)...}; return on_submit_fn<Fns...>{std::move(fns)...};
} }
template <class Fn>
struct on_now_fn : overload_fn<Fn> {
constexpr on_now_fn() = default;
using overload_fn<Fn>::overload_fn;
};
template <class Fn>
auto on_now(Fn fn) -> on_now_fn<Fn> {
return on_now_fn<Fn>{std::move(fn)};
}
} // namespace pushmi } // namespace pushmi
...@@ -27,6 +27,10 @@ struct receiver_category {}; ...@@ -27,6 +27,10 @@ struct receiver_category {};
struct sender_category {}; struct sender_category {};
// for senders that are executors
struct executor_category {};
// time and constrained are mutually exclusive refinements of sender (time is a special case of constrained and may be folded in later) // time and constrained are mutually exclusive refinements of sender (time is a special case of constrained and may be folded in later)
...@@ -149,6 +153,23 @@ PUSHMI_INLINE_VAR constexpr bool is_sender_v = is_sender<PS>::value; ...@@ -149,6 +153,23 @@ PUSHMI_INLINE_VAR constexpr bool is_sender_v = is_sender<PS>::value;
// is_sender_v<PS> // is_sender_v<PS>
// ); // );
// Executor trait and tag
template<class... TN>
struct is_executor;
// Tag
template<>
struct is_executor<> { using property_category = executor_category; };
// Trait
template<class PS>
struct is_executor<PS> : property_query<PS, is_executor<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_executor_v = is_executor<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept Executor,
is_executor_v<PS> && is_sender_v<PS> && is_single_v<PS>
);
// Time trait and tag // Time trait and tag
template<class... TN> template<class... TN>
struct is_time; struct is_time;
...@@ -163,7 +184,7 @@ PUSHMI_INLINE_VAR constexpr bool is_time_v = is_time<PS>::value; ...@@ -163,7 +184,7 @@ PUSHMI_INLINE_VAR constexpr bool is_time_v = is_time<PS>::value;
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class PS) template (class PS)
concept Time, concept Time,
is_time_v<PS> is_time_v<PS> && is_sender_v<PS>
); );
// Constrained trait and tag // Constrained trait and tag
...@@ -180,7 +201,7 @@ PUSHMI_INLINE_VAR constexpr bool is_constrained_v = is_constrained<PS>::value; ...@@ -180,7 +201,7 @@ PUSHMI_INLINE_VAR constexpr bool is_constrained_v = is_constrained<PS>::value;
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class PS) template (class PS)
concept Constrained, concept Constrained,
is_constrained_v<PS> is_constrained_v<PS> && is_sender_v<PS>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
...@@ -191,7 +212,8 @@ PUSHMI_CONCEPT_DEF( ...@@ -191,7 +212,8 @@ PUSHMI_CONCEPT_DEF(
) && ) &&
SemiMovable<S> && SemiMovable<S> &&
property_query_v<S, PropertyN...> && property_query_v<S, PropertyN...> &&
is_receiver_v<S> is_receiver_v<S> &&
!is_sender_v<S>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
...@@ -236,10 +258,15 @@ PUSHMI_CONCEPT_DEF( ...@@ -236,10 +258,15 @@ PUSHMI_CONCEPT_DEF(
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class D, class... PropertyN) template (class D, class... PropertyN)
(concept Sender)(D, PropertyN...), (concept Sender)(D, PropertyN...),
requires(D& d) (
::pushmi::executor(d),
requires_<Executor<decltype(::pushmi::executor(d))>>
) &&
SemiMovable<D> && SemiMovable<D> &&
None<D> && None<D> &&
property_query_v<D, PropertyN...> && property_query_v<D, PropertyN...> &&
is_sender_v<D> is_sender_v<D> &&
!is_receiver_v<D>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
...@@ -253,6 +280,11 @@ PUSHMI_CONCEPT_DEF( ...@@ -253,6 +280,11 @@ PUSHMI_CONCEPT_DEF(
property_query_v<D, PropertyN...> property_query_v<D, PropertyN...>
); );
template <class D>
PUSHMI_PP_CONSTRAINED_USING(
Sender<D>,
executor_t =, decltype(::pushmi::executor(std::declval<D&>())));
// add concepts to support cancellation // add concepts to support cancellation
// //
......
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
#include <chrono> #include <chrono>
#include <functional> #include <functional>
#include "time_single_sender.h" #include "single.h"
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
...@@ -29,7 +29,7 @@ private: ...@@ -29,7 +29,7 @@ private:
template <class T> template <class T>
using wrapped_t = detail::not_any_time_executor_ref_t<T>; using wrapped_t = detail::not_any_time_executor_ref_t<T>;
public: public:
using properties = property_set<is_time<>, is_single<>>; using properties = property_set<is_time<>, is_executor<>, is_single<>>;
any_time_executor_ref() = delete; any_time_executor_ref() = delete;
any_time_executor_ref(const any_time_executor_ref&) = default; any_time_executor_ref(const any_time_executor_ref&) = default;
...@@ -66,6 +66,7 @@ public: ...@@ -66,6 +66,7 @@ public:
std::chrono::system_clock::time_point now() { std::chrono::system_clock::time_point now() {
return vptr_->now_(pobj_); return vptr_->now_(pobj_);
} }
any_time_executor_ref executor() { return *this; }
template<class SingleReceiver> template<class SingleReceiver>
void submit(TP tp, SingleReceiver&& sa) { void submit(TP tp, SingleReceiver&& sa) {
// static_assert( // static_assert(
...@@ -122,7 +123,7 @@ using not_any_time_executor = ...@@ -122,7 +123,7 @@ using not_any_time_executor =
std::decay_t<T>>; std::decay_t<T>>;
} // namespace detail } // namespace detail
template<class E, class TP> template <class E, class TP>
struct any_time_executor : detail::any_time_executor_base<E, TP> { struct any_time_executor : detail::any_time_executor_base<E, TP> {
constexpr any_time_executor() = default; constexpr any_time_executor() = default;
using detail::any_time_executor_base<E, TP>::any_time_executor_base; using detail::any_time_executor_base<E, TP>::any_time_executor_base;
......
...@@ -11,6 +11,11 @@ ...@@ -11,6 +11,11 @@
namespace pushmi { namespace pushmi {
namespace __adl { namespace __adl {
//
// support methods on a class reference
//
PUSHMI_TEMPLATE (class S) PUSHMI_TEMPLATE (class S)
(requires requires (std::declval<S&>().done())) (requires requires (std::declval<S&>().done()))
void set_done(S& s) noexcept(noexcept(s.done())) { void set_done(S& s) noexcept(noexcept(s.done())) {
...@@ -38,6 +43,12 @@ void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) { ...@@ -38,6 +43,12 @@ void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) {
s.starting((Up&&) up); s.starting((Up&&) up);
} }
PUSHMI_TEMPLATE (class SD)
(requires requires (std::declval<SD&>().executor()))
auto executor(SD& sd) noexcept(noexcept(sd.executor())) {
return sd.executor();
}
PUSHMI_TEMPLATE (class SD, class Out) PUSHMI_TEMPLATE (class SD, class Out)
(requires requires (std::declval<SD&>().submit(std::declval<Out>()))) (requires requires (std::declval<SD&>().submit(std::declval<Out>())))
void submit(SD& sd, Out out) noexcept(noexcept(sd.submit(std::move(out)))) { void submit(SD& sd, Out out) noexcept(noexcept(sd.submit(std::move(out)))) {
...@@ -61,6 +72,70 @@ void submit(SD& sd, TP tp, Out out) ...@@ -61,6 +72,70 @@ void submit(SD& sd, TP tp, Out out)
sd.submit(std::move(tp), std::move(out)); sd.submit(std::move(tp), std::move(out));
} }
//
// support methods on a class pointer
//
PUSHMI_TEMPLATE (class S)
(requires requires (std::declval<S&>()->done()))
void set_done(S& s) noexcept(noexcept(s->done())) {
s->done();
}
PUSHMI_TEMPLATE (class S, class E)
(requires requires (std::declval<S&>()->error(std::declval<E>())))
void set_error(S& s, E e) noexcept(noexcept(s->error(std::move(e)))) {
s->error(std::move(e));
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>()->value(std::declval<V&&>())))
void set_value(S& s, V&& v) noexcept(noexcept(s->value((V&&) v))) {
s->value((V&&) v);
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>()->next(std::declval<V&&>())))
void set_next(S& s, V&& v) noexcept(noexcept(s->next((V&&) v))) {
s->next((V&&) v);
}
PUSHMI_TEMPLATE (class S, class Up)
(requires requires (std::declval<S&>()->starting(std::declval<Up&&>())))
void set_starting(S& s, Up&& up) noexcept(noexcept(s->starting((Up&&) up))) {
s->starting((Up&&) up);
}
PUSHMI_TEMPLATE (class SD)
(requires requires (std::declval<SD&>()->executor()))
auto executor(SD& sd) noexcept(noexcept(sd->executor())) {
return sd->executor();
}
PUSHMI_TEMPLATE (class SD, class Out)
(requires requires (std::declval<SD&>()->submit(std::declval<Out>())))
void submit(SD& sd, Out out) noexcept(noexcept(sd->submit(std::move(out)))) {
sd->submit(std::move(out));
}
PUSHMI_TEMPLATE (class SD)
(requires requires (std::declval<SD&>()->now()))
auto now(SD& sd) noexcept(noexcept(sd->now())) {
return sd->now();
}
PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires (
std::declval<SD&>()->submit(
std::declval<TP(&)(TP)>()(std::declval<SD&>()->now()),
std::declval<Out>())
))
void submit(SD& sd, TP tp, Out out)
noexcept(noexcept(sd->submit(std::move(tp), std::move(out)))) {
sd->submit(std::move(tp), std::move(out));
}
//
// add support for std::promise externally
//
template <class T> template <class T>
void set_done(std::promise<T>& p) noexcept( void set_done(std::promise<T>& p) noexcept(
noexcept(p.set_exception(std::make_exception_ptr(0)))) { noexcept(p.set_exception(std::make_exception_ptr(0)))) {
...@@ -83,6 +158,10 @@ void set_value(std::promise<T>& s, T t) { ...@@ -83,6 +158,10 @@ void set_value(std::promise<T>& s, T t) {
s.set_value(std::move(t)); s.set_value(std::move(t));
} }
//
// support reference_wrapper
//
PUSHMI_TEMPLATE (class S) PUSHMI_TEMPLATE (class S)
(requires requires ( set_done(std::declval<S&>()) )) (requires requires ( set_done(std::declval<S&>()) ))
void set_done(std::reference_wrapper<S> s) noexcept( void set_done(std::reference_wrapper<S> s) noexcept(
...@@ -112,6 +191,11 @@ void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept( ...@@ -112,6 +191,11 @@ void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept(
noexcept(set_starting(s.get(), (Up&&) up))) { noexcept(set_starting(s.get(), (Up&&) up))) {
set_starting(s.get(), (Up&&) up); set_starting(s.get(), (Up&&) up);
} }
PUSHMI_TEMPLATE (class SD)
(requires requires ( executor(std::declval<SD&>()) ))
auto executor(std::reference_wrapper<SD> sd) noexcept(noexcept(executor(sd.get()))) {
return executor(sd.get());
}
PUSHMI_TEMPLATE (class SD, class Out) PUSHMI_TEMPLATE (class SD, class Out)
(requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) )) (requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) ))
void submit(std::reference_wrapper<SD> sd, Out out) noexcept( void submit(std::reference_wrapper<SD> sd, Out out) noexcept(
...@@ -135,6 +219,9 @@ void submit(std::reference_wrapper<SD> sd, TP tp, Out out) ...@@ -135,6 +219,9 @@ void submit(std::reference_wrapper<SD> sd, TP tp, Out out)
submit(sd.get(), std::move(tp), std::move(out)); submit(sd.get(), std::move(tp), std::move(out));
} }
//
// accessors for free functions in this namespace
//
struct set_done_fn { struct set_done_fn {
PUSHMI_TEMPLATE (class S) PUSHMI_TEMPLATE (class S)
...@@ -207,6 +294,16 @@ struct set_starting_fn { ...@@ -207,6 +294,16 @@ struct set_starting_fn {
} }
}; };
struct get_executor_fn {
PUSHMI_TEMPLATE (class SD)
(requires requires (
executor(std::declval<SD&>())
))
auto operator()(SD&& sd) const noexcept(noexcept(executor(sd))) {
return executor(sd);
}
};
struct do_submit_fn { struct do_submit_fn {
PUSHMI_TEMPLATE (class SD, class Out) PUSHMI_TEMPLATE (class SD, class Out)
(requires requires ( (requires requires (
...@@ -247,6 +344,7 @@ PUSHMI_INLINE_VAR constexpr __adl::set_error_fn set_error{}; ...@@ -247,6 +344,7 @@ PUSHMI_INLINE_VAR constexpr __adl::set_error_fn set_error{};
PUSHMI_INLINE_VAR constexpr __adl::set_value_fn set_value{}; PUSHMI_INLINE_VAR constexpr __adl::set_value_fn set_value{};
PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{}; PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{};
PUSHMI_INLINE_VAR constexpr __adl::set_starting_fn set_starting{}; PUSHMI_INLINE_VAR constexpr __adl::set_starting_fn set_starting{};
PUSHMI_INLINE_VAR constexpr __adl::get_executor_fn executor{};
PUSHMI_INLINE_VAR constexpr __adl::do_submit_fn submit{}; PUSHMI_INLINE_VAR constexpr __adl::do_submit_fn submit{};
PUSHMI_INLINE_VAR constexpr __adl::get_now_fn now{}; PUSHMI_INLINE_VAR constexpr __adl::get_now_fn now{};
PUSHMI_INLINE_VAR constexpr __adl::get_now_fn top{}; PUSHMI_INLINE_VAR constexpr __adl::get_now_fn top{};
......
...@@ -235,7 +235,7 @@ class flow_many<Data, DNF, DEF, DDF, DStrtF> { ...@@ -235,7 +235,7 @@ class flow_many<Data, DNF, DEF, DDF, DStrtF> {
DStrtF strtf_; DStrtF strtf_;
public: public:
using properties = property_set<is_receiver<>, is_flow<>, is_many<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_receiver<>, is_flow<>, is_many<>>>;
static_assert( static_assert(
!detail::is_v<DNF, on_error_fn>, !detail::is_v<DNF, on_error_fn>,
......
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "flow_many.h" #include "flow_many.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
...@@ -21,8 +23,10 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -21,8 +23,10 @@ class flow_many_sender<V, PV, PE, E> {
} }
struct vtable { struct vtable {
static void s_op(data&, data*) {} static void s_op(data&, data*) {}
static any_time_executor<E /* hmm, TP will be invasive */> s_executor(data&) { return {}; }
static void s_submit(data&, any_flow_many<V, PV, PE, E>) {} static void s_submit(data&, any_flow_many<V, PV, PE, E>) {}
void (*op_)(data&, data*) = vtable::s_op; void (*op_)(data&, data*) = vtable::s_op;
any_time_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, any_flow_many<V, PV, PE, E>) = vtable::s_submit; void (*submit_)(data&, any_flow_many<V, PV, PE, E>) = vtable::s_submit;
}; };
static constexpr vtable const noop_ {}; static constexpr vtable const noop_ {};
...@@ -35,11 +39,14 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -35,11 +39,14 @@ class flow_many_sender<V, PV, PE, E> {
dst->pobj_ = std::exchange(src.pobj_, nullptr); dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_); delete static_cast<Wrapped const*>(src.pobj_);
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) { static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out)); ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
data_.pobj_ = new Wrapped(std::move(obj)); data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -53,13 +60,16 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -53,13 +60,16 @@ class flow_many_sender<V, PV, PE, E> {
std::move(*static_cast<Wrapped*>((void*)src.buffer_))); std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped(); static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) { static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
::pushmi::submit( ::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), *static_cast<Wrapped*>((void*)src.buffer_),
std::move(out)); std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
new (data_.buffer_) Wrapped(std::move(obj)); new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -87,6 +97,9 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -87,6 +97,9 @@ class flow_many_sender<V, PV, PE, E> {
new ((void*)this) flow_many_sender(std::move(that)); new ((void*)this) flow_many_sender(std::move(that));
return *this; return *this;
} }
any_time_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(any_flow_many<V, PV, PE, E> out) { void submit(any_flow_many<V, PV, PE, E> out) {
vptr_->submit_(data_, std::move(out)); vptr_->submit_(data_, std::move(out));
} }
...@@ -97,9 +110,10 @@ template <class V, class PV, class PE, class E> ...@@ -97,9 +110,10 @@ template <class V, class PV, class PE, class E>
constexpr typename flow_many_sender<V, PV, PE, E>::vtable const constexpr typename flow_many_sender<V, PV, PE, E>::vtable const
flow_many_sender<V, PV, PE, E>::noop_; flow_many_sender<V, PV, PE, E>::noop_;
template <class SF> template <class SF, class EXF>
class flow_many_sender<SF> { class flow_many_sender<SF, EXF> {
SF sf_; SF sf_;
EXF exf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>; using properties = property_set<is_sender<>, is_flow<>, is_many<>>;
...@@ -107,7 +121,10 @@ class flow_many_sender<SF> { ...@@ -107,7 +121,10 @@ class flow_many_sender<SF> {
constexpr flow_many_sender() = default; constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(SF sf) constexpr explicit flow_many_sender(SF sf)
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
constexpr flow_many_sender(SF sf, EXF exf)
: sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_many<>, is_flow<>> && Invocable<SF&, Out>) (requires Receiver<Out, is_many<>, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
...@@ -115,19 +132,24 @@ class flow_many_sender<SF> { ...@@ -115,19 +132,24 @@ class flow_many_sender<SF> {
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_many<>, is_flow<>>) Data, class DSF> template <PUSHMI_TYPE_CONSTRAINT(Sender<is_many<>, is_flow<>>) Data, class DSF, class DEXF>
class flow_many_sender<Data, DSF> { class flow_many_sender<Data, DSF, DEXF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
DEXF exf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_flow<>, is_many<>>>;
constexpr flow_many_sender() = default; constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(Data data) constexpr explicit flow_many_sender(Data data)
: data_(std::move(data)) {} : data_(std::move(data)) {}
constexpr flow_many_sender(Data data, DSF sf) constexpr flow_many_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
constexpr flow_many_sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>, is_flow<>> PUSHMI_AND (requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>, is_flow<>> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>)) lazy::Invocable<DSF&, Data&, Out>))
...@@ -140,41 +162,59 @@ class flow_many_sender<Data, DSF> { ...@@ -140,41 +162,59 @@ class flow_many_sender<Data, DSF> {
// make_flow_many_sender // make_flow_many_sender
PUSHMI_INLINE_VAR constexpr struct make_flow_many_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_flow_many_sender_fn {
inline auto operator()() const { inline auto operator()() const {
return flow_many_sender<ignoreSF>{}; return flow_many_sender<ignoreSF, trampolineEXF>{};
} }
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const { auto operator()(SF sf) const {
return flow_many_sender<SF>{std::move(sf)}; return flow_many_sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return flow_many_sender<SF, EXF>{std::move(sf), std::move(exf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>, is_flow<>>) (requires True<> && Sender<Data, is_many<>, is_flow<>>)
auto operator()(Data d) const { auto operator()(Data d) const {
return flow_many_sender<Data, passDSF>{std::move(d)}; return flow_many_sender<Data, passDSF, passDEXF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>, is_flow<>>) (requires Sender<Data, is_many<>, is_flow<>>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return flow_many_sender<Data, DSF>{std::move(d), std::move(sf)}; return flow_many_sender<Data, DSF, passDEXF>{std::move(d), std::move(sf)};
}
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_many<>, is_flow<>> && Invocable<DEXF&, Data&>)
auto operator()(Data d, DSF sf, DEXF exf) const {
return flow_many_sender<Data, DSF, DEXF>{std::move(d), std::move(sf), std::move(exf)};
} }
} const make_flow_many_sender {}; } const make_flow_many_sender {};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// deduction guides // deduction guides
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
flow_many_sender() -> flow_many_sender<ignoreSF>; flow_many_sender() -> flow_many_sender<ignoreSF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_many_sender(SF) -> flow_many_sender<SF>; flow_many_sender(SF) -> flow_many_sender<SF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_many_sender(SF, EXF) -> flow_many_sender<SF, EXF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>, is_flow<>>) (requires True<> && Sender<Data, is_many<>, is_flow<>>)
flow_many_sender(Data) -> flow_many_sender<Data, passDSF>; flow_many_sender(Data) -> flow_many_sender<Data, passDSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>, is_flow<>>) (requires Sender<Data, is_many<>, is_flow<>>)
flow_many_sender(Data, DSF) -> flow_many_sender<Data, DSF>; flow_many_sender(Data, DSF) -> flow_many_sender<Data, DSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_many<>, is_flow<>> && Invocable<DEXF&, Data&>)
flow_many_sender(Data, DSF, DEXF) -> flow_many_sender<Data, DSF, DEXF>;
#endif #endif
template <class V, class PV = std::ptrdiff_t, class PE = std::exception_ptr, class E = PE> template <class V, class PV = std::ptrdiff_t, class PE = std::exception_ptr, class E = PE>
......
...@@ -205,7 +205,7 @@ class flow_single<Data, DVF, DEF, DDF, DStrtF> { ...@@ -205,7 +205,7 @@ class flow_single<Data, DVF, DEF, DDF, DStrtF> {
DStrtF strtf_; DStrtF strtf_;
public: public:
using properties = property_set<is_receiver<>, is_flow<>, is_single<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_receiver<>, is_flow<>, is_single<>>>;
static_assert( static_assert(
!detail::is_v<DVF, on_error_fn>, !detail::is_v<DVF, on_error_fn>,
......
...@@ -5,11 +5,13 @@ ...@@ -5,11 +5,13 @@
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "flow_single.h" #include "flow_single.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
template <class V, class PE, class E> template <class V, class PE = std::exception_ptr, class E = PE>
class flow_single_sender<V, PE, E> { class any_flow_single_sender {
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
char buffer_[sizeof(V)]; // can hold a V in-situ char buffer_[sizeof(V)]; // can hold a V in-situ
...@@ -21,31 +23,36 @@ class flow_single_sender<V, PE, E> { ...@@ -21,31 +23,36 @@ class flow_single_sender<V, PE, E> {
} }
struct vtable { struct vtable {
static void s_op(data&, data*) {} static void s_op(data&, data*) {}
static any_time_executor<E /* hmm, TP will be invasive */> s_executor(data&) { return {}; }
static void s_submit(data&, flow_single<V, PE, E>) {} static void s_submit(data&, flow_single<V, PE, E>) {}
void (*op_)(data&, data*) = vtable::s_op; void (*op_)(data&, data*) = vtable::s_op;
any_time_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, flow_single<V, PE, E>) = vtable::s_submit; void (*submit_)(data&, flow_single<V, PE, E>) = vtable::s_submit;
}; };
static constexpr vtable const noop_ {}; static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_; vtable const* vptr_ = &noop_;
template <class Wrapped> template <class Wrapped>
flow_single_sender(Wrapped obj, std::false_type) : flow_single_sender() { any_flow_single_sender(Wrapped obj, std::false_type) : any_flow_single_sender() {
struct s { struct s {
static void op(data& src, data* dst) { static void op(data& src, data* dst) {
if (dst) if (dst)
dst->pobj_ = std::exchange(src.pobj_, nullptr); dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_); delete static_cast<Wrapped const*>(src.pobj_);
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, flow_single<V, PE, E> out) { static void submit(data& src, flow_single<V, PE, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out)); ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
data_.pobj_ = new Wrapped(std::move(obj)); data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
template <class Wrapped> template <class Wrapped>
flow_single_sender(Wrapped obj, std::true_type) noexcept any_flow_single_sender(Wrapped obj, std::true_type) noexcept
: flow_single_sender() { : any_flow_single_sender() {
struct s { struct s {
static void op(data& src, data* dst) { static void op(data& src, data* dst) {
if (dst) if (dst)
...@@ -53,40 +60,46 @@ class flow_single_sender<V, PE, E> { ...@@ -53,40 +60,46 @@ class flow_single_sender<V, PE, E> {
std::move(*static_cast<Wrapped*>((void*)src.buffer_))); std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped(); static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, flow_single<V, PE, E> out) { static void submit(data& src, flow_single<V, PE, E> out) {
::pushmi::submit( ::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), *static_cast<Wrapped*>((void*)src.buffer_),
std::move(out)); std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
new (data_.buffer_) Wrapped(std::move(obj)); new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
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<U, flow_single_sender>::value, U>; std::enable_if_t<!std::is_same<U, any_flow_single_sender>::value, U>;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_single<>>; using properties = property_set<is_sender<>, is_flow<>, is_single<>>;
flow_single_sender() = default; any_flow_single_sender() = default;
flow_single_sender(flow_single_sender&& that) noexcept any_flow_single_sender(any_flow_single_sender&& that) noexcept
: flow_single_sender() { : any_flow_single_sender() {
that.vptr_->op_(that.data_, &data_); that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_); std::swap(that.vptr_, vptr_);
} }
PUSHMI_TEMPLATE (class Wrapped) PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>, is_single<>>) (requires FlowSender<wrapped_t<Wrapped>, is_single<>>)
explicit flow_single_sender(Wrapped obj) noexcept(insitu<Wrapped>()) explicit any_flow_single_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: flow_single_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {} : any_flow_single_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~flow_single_sender() { ~any_flow_single_sender() {
vptr_->op_(data_, nullptr); vptr_->op_(data_, nullptr);
} }
flow_single_sender& operator=(flow_single_sender&& that) noexcept { any_flow_single_sender& operator=(any_flow_single_sender&& that) noexcept {
this->~flow_single_sender(); this->~any_flow_single_sender();
new ((void*)this) flow_single_sender(std::move(that)); new ((void*)this) any_flow_single_sender(std::move(that));
return *this; return *this;
} }
any_time_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(flow_single<V, PE, E> out) { void submit(flow_single<V, PE, E> out) {
vptr_->submit_(data_, std::move(out)); vptr_->submit_(data_, std::move(out));
} }
...@@ -94,12 +107,13 @@ class flow_single_sender<V, PE, E> { ...@@ -94,12 +107,13 @@ class flow_single_sender<V, PE, E> {
// Class static definitions: // Class static definitions:
template <class V, class PE, class E> template <class V, class PE, class E>
constexpr typename flow_single_sender<V, PE, E>::vtable const constexpr typename any_flow_single_sender<V, PE, E>::vtable const
flow_single_sender<V, PE, E>::noop_; any_flow_single_sender<V, PE, E>::noop_;
template <class SF> template <class SF, class EXF>
class flow_single_sender<SF> { class flow_single_sender<SF, EXF> {
SF sf_; SF sf_;
EXF exf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_single<>>; using properties = property_set<is_sender<>, is_flow<>, is_single<>>;
...@@ -107,7 +121,10 @@ class flow_single_sender<SF> { ...@@ -107,7 +121,10 @@ class flow_single_sender<SF> {
constexpr flow_single_sender() = default; constexpr flow_single_sender() = default;
constexpr explicit flow_single_sender(SF sf) constexpr explicit flow_single_sender(SF sf)
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
constexpr flow_single_sender(SF sf, EXF exf)
: sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_single<>, is_flow<>> && Invocable<SF&, Out>) (requires Receiver<Out, is_single<>, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
...@@ -115,19 +132,24 @@ class flow_single_sender<SF> { ...@@ -115,19 +132,24 @@ class flow_single_sender<SF> {
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data, class DSF> template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data, class DSF, class DEXF>
class flow_single_sender<Data, DSF> { class flow_single_sender<Data, DSF, DEXF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
DEXF exf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_single<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_flow<>, is_single<>>>;
constexpr flow_single_sender() = default; constexpr flow_single_sender() = default;
constexpr explicit flow_single_sender(Data data) constexpr explicit flow_single_sender(Data data)
: data_(std::move(data)) {} : data_(std::move(data)) {}
constexpr flow_single_sender(Data data, DSF sf) constexpr flow_single_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
constexpr flow_single_sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>, is_flow<>> PUSHMI_AND (requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>, is_flow<>> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>)) lazy::Invocable<DSF&, Data&, Out>))
...@@ -140,45 +162,60 @@ class flow_single_sender<Data, DSF> { ...@@ -140,45 +162,60 @@ class flow_single_sender<Data, DSF> {
// make_flow_single_sender // make_flow_single_sender
PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn {
inline auto operator()() const { inline auto operator()() const {
return flow_single_sender<ignoreSF>{}; return flow_single_sender<ignoreSF, trampolineEXF>{};
} }
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const { auto operator()(SF sf) const {
return flow_single_sender<SF>{std::move(sf)}; return flow_single_sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return flow_single_sender<SF, EXF>{std::move(sf), std::move(exf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>) (requires True<> && Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d) const { auto operator()(Data d) const {
return flow_single_sender<Data, passDSF>{std::move(d)}; return flow_single_sender<Data, passDSF, passDEXF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>) (requires Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return flow_single_sender<Data, DSF>{std::move(d), std::move(sf)}; return flow_single_sender<Data, DSF, passDEXF>{std::move(d), std::move(sf)};
}
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_single<>, is_flow<>> && Invocable<DEXF&, Data&>)
auto operator()(Data d, DSF sf, DEXF exf) const {
return flow_single_sender<Data, DSF, DEXF>{std::move(d), std::move(sf), std::move(exf)};
} }
} const make_flow_single_sender {}; } const make_flow_single_sender {};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// deduction guides // deduction guides
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
flow_single_sender() -> flow_single_sender<ignoreSF>; flow_single_sender() -> flow_single_sender<ignoreSF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_single_sender(SF) -> flow_single_sender<SF>; flow_single_sender(SF) -> flow_single_sender<SF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_single_sender(SF, EXF) -> flow_single_sender<SF, EXF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>) (requires True<> && Sender<Data, is_single<>, is_flow<>>)
flow_single_sender(Data) -> flow_single_sender<Data, passDSF>; flow_single_sender(Data) -> flow_single_sender<Data, passDSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>) (requires Sender<Data, is_single<>, is_flow<>>)
flow_single_sender(Data, DSF) -> flow_single_sender<Data, DSF>; flow_single_sender(Data, DSF) -> flow_single_sender<Data, DSF, passDEXF>;
#endif
template <class V, class PE = std::exception_ptr, class E = PE> PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
using any_flow_single_sender = flow_single_sender<V, PE, E>; (requires Sender<Data, is_single<>, is_flow<>> && Invocable<DEXF&, Data&>)
flow_single_sender(Data, DSF, DEXF) -> flow_single_sender<Data, DSF, DEXF>;
#endif
template<> template<>
struct construct_deduced<flow_single_sender> struct construct_deduced<flow_single_sender>
......
...@@ -80,6 +80,17 @@ class flow_many; ...@@ -80,6 +80,17 @@ class flow_many;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many_sender; class flow_many_sender;
template<
class V,
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
struct any_time_single_sender;
template<
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
struct any_time_executor;
template< template<
class E = std::exception_ptr, class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point> class TP = std::chrono::system_clock::time_point>
......
#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 "executor.h"
namespace pushmi {
class inline_time_executor {
public:
using properties = property_set<is_time<>, is_executor<>, is_single<>>;
auto now() {
return std::chrono::system_clock::now();
}
auto executor() { return *this; }
PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out, is_single<>>)
void submit(TP tp, Out out) {
std::this_thread::sleep_until(tp);
::pushmi::set_value(std::move(out), *this);
}
};
struct inlineEXF {
inline_time_executor operator()(){
return {};
}
};
inline inline_time_executor inline_executor() {
return {};
}
} // namespace pushmi
...@@ -229,7 +229,7 @@ class many<Data, DNF, DEF, DDF> { ...@@ -229,7 +229,7 @@ class many<Data, DNF, DEF, DDF> {
"error function must be noexcept and support std::exception_ptr"); "error function must be noexcept and support std::exception_ptr");
public: public:
using properties = property_set<is_receiver<>, is_many<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_receiver<>, is_many<>>>;
constexpr explicit many(Data d) constexpr explicit many(Data d)
: many(std::move(d), DNF{}, DEF{}, DDF{}) {} : many(std::move(d), DNF{}, DEF{}, DDF{}) {}
......
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "many.h" #include "many.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
...@@ -21,8 +23,10 @@ class any_many_sender { ...@@ -21,8 +23,10 @@ class any_many_sender {
} }
struct vtable { struct vtable {
static void s_op(data&, data*) {} static void s_op(data&, data*) {}
static any_time_executor<E /* hmm, TP will be invasive */> s_executor(data&) { return {}; }
static void s_submit(data&, many<V, E>) {} static void s_submit(data&, many<V, E>) {}
void (*op_)(data&, data*) = vtable::s_op; void (*op_)(data&, data*) = vtable::s_op;
any_time_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, many<V, E>) = vtable::s_submit; void (*submit_)(data&, many<V, E>) = vtable::s_submit;
}; };
static constexpr vtable const noop_ {}; static constexpr vtable const noop_ {};
...@@ -35,11 +39,14 @@ class any_many_sender { ...@@ -35,11 +39,14 @@ class any_many_sender {
dst->pobj_ = std::exchange(src.pobj_, nullptr); dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_); delete static_cast<Wrapped const*>(src.pobj_);
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, many<V, E> out) { static void submit(data& src, many<V, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out)); ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
data_.pobj_ = new Wrapped(std::move(obj)); data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -53,12 +60,15 @@ class any_many_sender { ...@@ -53,12 +60,15 @@ class any_many_sender {
std::move(*static_cast<Wrapped*>((void*)src.buffer_))); std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped(); static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, many<V, E> out) { static void submit(data& src, many<V, E> out) {
::pushmi::submit( ::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out)); *static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
new (data_.buffer_) Wrapped(std::move(obj)); new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -87,6 +97,9 @@ class any_many_sender { ...@@ -87,6 +97,9 @@ class any_many_sender {
new ((void*)this) any_many_sender(std::move(that)); new ((void*)this) any_many_sender(std::move(that));
return *this; return *this;
} }
any_time_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(many<V, E> out) { void submit(many<V, E> out) {
vptr_->submit_(data_, std::move(out)); vptr_->submit_(data_, std::move(out));
} }
...@@ -97,9 +110,10 @@ template <class V, class E> ...@@ -97,9 +110,10 @@ template <class V, class E>
constexpr typename any_many_sender<V, E>::vtable const constexpr typename any_many_sender<V, E>::vtable const
any_many_sender<V, E>::noop_; any_many_sender<V, E>::noop_;
template <class SF> template <class SF, class EXF>
class many_sender<SF> { class many_sender<SF, EXF> {
SF sf_; SF sf_;
EXF exf_;
public: public:
using properties = property_set<is_sender<>, is_many<>>; using properties = property_set<is_sender<>, is_many<>>;
...@@ -107,7 +121,10 @@ class many_sender<SF> { ...@@ -107,7 +121,10 @@ class many_sender<SF> {
constexpr many_sender() = default; constexpr many_sender() = default;
constexpr explicit many_sender(SF sf) constexpr explicit many_sender(SF sf)
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
constexpr many_sender(SF sf, EXF exf)
: sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND lazy::Invocable<SF&, Out>)) (requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND lazy::Invocable<SF&, Out>))
void submit(Out out) { void submit(Out out) {
...@@ -115,20 +132,24 @@ class many_sender<SF> { ...@@ -115,20 +132,24 @@ class many_sender<SF> {
} }
}; };
namespace detail { template <PUSHMI_TYPE_CONSTRAINT(Sender<is_many<>>) Data, class DSF, class DEXF>
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_many<>>) Data, class DSF> class many_sender<Data, DSF, DEXF> {
class many_sender_2 {
Data data_; Data data_;
DSF sf_; DSF sf_;
DEXF exf_;
public: public:
using properties = property_set<is_sender<>, is_many<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_many<>>>;
constexpr many_sender_2() = default; constexpr many_sender() = default;
constexpr explicit many_sender_2(Data data) constexpr explicit many_sender(Data data)
: data_(std::move(data)) {} : data_(std::move(data)) {}
constexpr many_sender_2(Data data, DSF sf) constexpr many_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
constexpr many_sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND (requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>)) lazy::Invocable<DSF&, Data&, Out>))
...@@ -137,60 +158,63 @@ class many_sender_2 { ...@@ -137,60 +158,63 @@ class many_sender_2 {
} }
}; };
template <class A, class B>
using many_sender_base =
std::conditional_t<
(bool)Sender<A, is_many<>>,
many_sender_2<A, B>,
any_many_sender<A, B>>;
} // namespace detail
template <class A, class B>
struct many_sender<A, B>
: detail::many_sender_base<A, B> {
constexpr many_sender() = default;
using detail::many_sender_base<A, B>::many_sender_base;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_many_sender // make_many_sender
PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn {
inline auto operator()() const { inline auto operator()() const {
return many_sender<ignoreSF>{}; return many_sender<ignoreSF, trampolineEXF>{};
} }
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const { auto operator()(SF sf) const {
return many_sender<SF>{std::move(sf)}; return many_sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return many_sender<SF, EXF>{std::move(sf), std::move(exf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>>) (requires True<> && Sender<Data, is_many<>>)
auto operator()(Data d) const { auto operator()(Data d) const {
return many_sender<Data, passDSF>{std::move(d)}; return many_sender<Data, passDSF, passDEXF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>>) (requires Sender<Data, is_many<>>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return many_sender<Data, DSF>{std::move(d), std::move(sf)}; return many_sender<Data, DSF, passDEXF>{std::move(d), std::move(sf)};
}
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_many<>> && Invocable<DEXF&, Data&>)
auto operator()(Data d, DSF sf, DEXF exf) const {
return many_sender<Data, DSF, DEXF>{std::move(d), std::move(sf), std::move(exf)};
} }
} const make_many_sender {}; } const make_many_sender {};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// deduction guides // deduction guides
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
many_sender() -> many_sender<ignoreSF>; many_sender() -> many_sender<ignoreSF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
many_sender(SF) -> many_sender<SF>; many_sender(SF) -> many_sender<SF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
many_sender(SF, EXF) -> many_sender<SF, EXF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>>) (requires True<> && Sender<Data, is_many<>>)
many_sender(Data) -> many_sender<Data, passDSF>; many_sender(Data) -> many_sender<Data, passDSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>>) (requires Sender<Data, is_many<>>)
many_sender(Data, DSF) -> many_sender<Data, DSF>; many_sender(Data, DSF) -> many_sender<Data, DSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_many<>> && Invocable<DEXF&, Data&>)
many_sender(Data, DSF, DEXF) -> many_sender<Data, DSF, DEXF>;
#endif #endif
template<> template<>
......
...@@ -12,12 +12,18 @@ namespace pushmi { ...@@ -12,12 +12,18 @@ namespace pushmi {
// very poor perf example executor. // very poor perf example executor.
// //
struct __new_thread_submit { struct new_thread_time_executor {
using properties = property_set<is_time<>, is_executor<>, is_single<>>;
auto now() {
return std::chrono::system_clock::now();
}
new_thread_time_executor executor() { return {}; }
PUSHMI_TEMPLATE(class TP, class Out) PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out>) (requires Regular<TP> && Receiver<Out>)
void operator()(TP at, Out out) const { void submit(TP at, Out out) {
std::thread t{[at = std::move(at), out = std::move(out)]() mutable { std::thread t{[at = std::move(at), out = std::move(out)]() mutable {
auto tr = trampoline(); auto tr = ::pushmi::trampoline();
::pushmi::submit(tr, std::move(at), std::move(out)); ::pushmi::submit(tr, std::move(at), std::move(out));
}}; }};
// pass ownership of thread to out // pass ownership of thread to out
...@@ -25,8 +31,8 @@ struct __new_thread_submit { ...@@ -25,8 +31,8 @@ struct __new_thread_submit {
} }
}; };
inline auto new_thread() { inline new_thread_time_executor new_thread() {
return make_time_single_sender(__new_thread_submit{}); return {};
} }
} } // namespace pushmi
...@@ -170,7 +170,7 @@ class none<Data, DEF, DDF> { ...@@ -170,7 +170,7 @@ class none<Data, DEF, DDF> {
!detail::is_v<Data, single>, !detail::is_v<Data, single>,
"none should not be used to wrap a single<>"); "none should not be used to wrap a single<>");
public: public:
using properties = property_set<is_receiver<>, is_none<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_receiver<>, is_none<>>>;
constexpr explicit none(Data d) : none(std::move(d), DEF{}, DDF{}) {} constexpr explicit none(Data d) : none(std::move(d), DEF{}, DDF{}) {}
constexpr none(Data d, DDF df) constexpr none(Data d, DDF df)
......
...@@ -294,6 +294,21 @@ public: ...@@ -294,6 +294,21 @@ public:
} }
}; };
struct executor_fn {
private:
struct impl {
PUSHMI_TEMPLATE (class In)
(requires Sender<In>)
auto operator()(In in) const {
return ::pushmi::executor(in);
}
};
public:
auto operator()() const {
return impl{};
}
};
struct do_submit_fn { struct do_submit_fn {
private: private:
template <class Out> template <class Out>
...@@ -352,6 +367,7 @@ PUSHMI_INLINE_VAR constexpr detail::set_error_fn set_error{}; ...@@ -352,6 +367,7 @@ PUSHMI_INLINE_VAR constexpr detail::set_error_fn set_error{};
PUSHMI_INLINE_VAR constexpr detail::set_value_fn set_value{}; PUSHMI_INLINE_VAR constexpr detail::set_value_fn set_value{};
PUSHMI_INLINE_VAR constexpr detail::set_next_fn set_next{}; PUSHMI_INLINE_VAR constexpr detail::set_next_fn set_next{};
PUSHMI_INLINE_VAR constexpr detail::set_starting_fn set_starting{}; PUSHMI_INLINE_VAR constexpr detail::set_starting_fn set_starting{};
PUSHMI_INLINE_VAR constexpr detail::executor_fn executor{};
PUSHMI_INLINE_VAR constexpr detail::do_submit_fn submit{}; PUSHMI_INLINE_VAR constexpr detail::do_submit_fn submit{};
PUSHMI_INLINE_VAR constexpr detail::now_fn now{}; PUSHMI_INLINE_VAR constexpr detail::now_fn now{};
PUSHMI_INLINE_VAR constexpr detail::now_fn top{}; PUSHMI_INLINE_VAR constexpr detail::now_fn top{};
......
...@@ -89,6 +89,14 @@ private: ...@@ -89,6 +89,14 @@ private:
} }
}; };
template <class In, class ExecutorFactory> template <class In, class ExecutorFactory>
struct executor_impl {
ExecutorFactory ef_;
template <class Data>
auto operator()(Data& data) const {
return ef_();
}
};
template <class In, class ExecutorFactory>
struct out_impl { struct out_impl {
ExecutorFactory ef_; ExecutorFactory ef_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
...@@ -113,7 +121,8 @@ private: ...@@ -113,7 +121,8 @@ private:
std::move(in), std::move(in),
::pushmi::detail::submit_transform_out<In>( ::pushmi::detail::submit_transform_out<In>(
out_impl<In, ExecutorFactory>{ef_} out_impl<In, ExecutorFactory>{ef_}
) ),
::pushmi::on_executor(executor_impl<In, ExecutorFactory>{ef_})
); );
} }
}; };
......
...@@ -140,10 +140,6 @@ using property_set_index_t = ...@@ -140,10 +140,6 @@ using property_set_index_t =
PropertySet<PS> && Property<P>, PropertySet<PS> && Property<P>,
decltype(detail::__property_set_index_fn<P>(PS{}))>; decltype(detail::__property_set_index_fn<P>(PS{}))>;
template <class T, class P>
using property_from_category_t =
property_set_index_t<properties_t<T>, P>;
template <class PS0, class PS1> template <class PS0, class PS1>
using property_set_insert_t = using property_set_insert_t =
typename std::enable_if_t< typename std::enable_if_t<
......
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "none.h" #include "none.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
...@@ -24,8 +26,10 @@ class sender<detail::erase_sender_t, E> { ...@@ -24,8 +26,10 @@ class sender<detail::erase_sender_t, E> {
} }
struct vtable { struct vtable {
static void s_op(data&, data*) {} static void s_op(data&, data*) {}
static any_time_executor<E /* hmm, TP will be invasive */> s_executor(data&) { return {}; }
static void s_submit(data&, any_none<E>) {} static void s_submit(data&, any_none<E>) {}
void (*op_)(data&, data*) = vtable::s_op; void (*op_)(data&, data*) = vtable::s_op;
any_time_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, any_none<E>) = vtable::s_submit; void (*submit_)(data&, any_none<E>) = vtable::s_submit;
}; };
static constexpr vtable const noop_ {}; static constexpr vtable const noop_ {};
...@@ -38,11 +42,14 @@ class sender<detail::erase_sender_t, E> { ...@@ -38,11 +42,14 @@ class sender<detail::erase_sender_t, E> {
dst->pobj_ = std::exchange(src.pobj_, nullptr); dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_); delete static_cast<Wrapped const*>(src.pobj_);
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, any_none<E> out) { static void submit(data& src, any_none<E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out)); ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
data_.pobj_ = new Wrapped(std::move(obj)); data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -55,12 +62,15 @@ class sender<detail::erase_sender_t, E> { ...@@ -55,12 +62,15 @@ class sender<detail::erase_sender_t, E> {
std::move(*static_cast<Wrapped*>((void*)src.buffer_))); std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped(); static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, any_none<E> out) { static void submit(data& src, any_none<E> out) {
::pushmi::submit( ::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out)); *static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
new (data_.buffer_) Wrapped(std::move(obj)); new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -87,6 +97,9 @@ class sender<detail::erase_sender_t, E> { ...@@ -87,6 +97,9 @@ class sender<detail::erase_sender_t, E> {
new ((void*)this) sender(std::move(that)); new ((void*)this) sender(std::move(that));
return *this; return *this;
} }
any_time_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(any_none<E> out) { void submit(any_none<E> out) {
vptr_->submit_(data_, std::move(out)); vptr_->submit_(data_, std::move(out));
} }
...@@ -97,15 +110,19 @@ template <class E> ...@@ -97,15 +110,19 @@ template <class E>
constexpr typename sender<detail::erase_sender_t, E>::vtable const constexpr typename sender<detail::erase_sender_t, E>::vtable const
sender<detail::erase_sender_t, E>::noop_; sender<detail::erase_sender_t, E>::noop_;
template <class SF> template <class SF, class EXF>
class sender<SF> { class sender<SF, EXF> {
SF sf_; SF sf_;
EXF exf_;
public: public:
using properties = property_set<is_sender<>, is_none<>>; using properties = property_set<is_sender<>, is_none<>>;
constexpr sender() = default; constexpr sender() = default;
constexpr explicit sender(SF sf) : sf_(std::move(sf)) {} constexpr explicit sender(SF sf) : sf_(std::move(sf)) {}
constexpr sender(SF sf, EXF exf) : sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_none<>> && Invocable<SF&, Out>) (requires Receiver<Out, is_none<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
...@@ -113,22 +130,26 @@ class sender<SF> { ...@@ -113,22 +130,26 @@ class sender<SF> {
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_none<>>) Data, class DSF> template <PUSHMI_TYPE_CONSTRAINT(Sender<is_none<>>) Data, class DSF, class DEXF>
class sender<Data, DSF> { class sender<Data, DSF, DEXF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
DEXF exf_;
static_assert(Sender<Data, is_none<>>, "The Data template parameter " static_assert(Sender<Data, is_none<>>, "The Data template parameter "
"must satisfy the Sender concept."); "must satisfy the Sender concept.");
public: public:
using properties = property_set<is_sender<>, is_none<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_none<>>>;
constexpr sender() = default; constexpr sender() = default;
constexpr explicit sender(Data data) constexpr explicit sender(Data data)
: data_(std::move(data)) {} : data_(std::move(data)) {}
constexpr sender(Data data, DSF sf) constexpr sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
constexpr sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_none<>> && Invocable<DSF&, Data&, Out>) (requires Receiver<Out, is_none<>> && Invocable<DSF&, Data&, Out>)
void submit(Out out) { void submit(Out out) {
...@@ -140,11 +161,17 @@ class sender<Data, DSF> { ...@@ -140,11 +161,17 @@ class sender<Data, DSF> {
// make_sender // make_sender
PUSHMI_INLINE_VAR constexpr struct make_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_sender_fn {
inline auto operator()() const { inline auto operator()() const {
return sender<ignoreSF>{}; return sender<ignoreSF, trampolineEXF>{};
} }
template <class SF> PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const { auto operator()(SF sf) const {
return sender<SF>{std::move(sf)}; return sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return sender<SF, EXF>{std::move(sf), std::move(exf)};
} }
PUSHMI_TEMPLATE(class Wrapped) PUSHMI_TEMPLATE(class Wrapped)
(requires Sender<Wrapped, is_none<>>) (requires Sender<Wrapped, is_none<>>)
...@@ -154,17 +181,27 @@ PUSHMI_INLINE_VAR constexpr struct make_sender_fn { ...@@ -154,17 +181,27 @@ PUSHMI_INLINE_VAR constexpr struct make_sender_fn {
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_none<>>) (requires Sender<Data, is_none<>>)
auto operator()(Data data, DSF sf) const { auto operator()(Data data, DSF sf) const {
return sender<Data, DSF>{std::move(data), std::move(sf)}; return sender<Data, DSF, passDEXF>{std::move(data), std::move(sf)};
}
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_none<>>)
auto operator()(Data data, DSF sf, DEXF exf) const {
return sender<Data, DSF, DEXF>{std::move(data), std::move(sf), std::move(exf)};
} }
} const make_sender {}; } const make_sender {};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// deduction guides // deduction guides
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
sender() -> sender<ignoreSF>; sender() -> sender<ignoreSF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
sender(SF) -> sender<SF, trampolineEXF>;
template <class SF> PUSHMI_TEMPLATE(class SF, class EXF)
sender(SF) -> sender<SF>; (requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
sender(SF, EXF) -> sender<SF, EXF>;
PUSHMI_TEMPLATE(class Wrapped) PUSHMI_TEMPLATE(class Wrapped)
(requires Sender<Wrapped, is_none<>>) (requires Sender<Wrapped, is_none<>>)
...@@ -173,7 +210,11 @@ sender(Wrapped) -> ...@@ -173,7 +210,11 @@ sender(Wrapped) ->
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_none<>>) (requires Sender<Data, is_none<>>)
sender(Data, DSF) -> sender<Data, DSF>; sender(Data, DSF) -> sender<Data, DSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_none<>>)
sender(Data, DSF, DEXF) -> sender<Data, DSF, DEXF>;
#endif #endif
template <class E = std::exception_ptr> template <class E = std::exception_ptr>
......
...@@ -232,7 +232,7 @@ class single<Data, DVF, DEF, DDF> { ...@@ -232,7 +232,7 @@ class single<Data, DVF, DEF, DDF> {
"error function must be noexcept and support std::exception_ptr"); "error function must be noexcept and support std::exception_ptr");
public: public:
using properties = property_set<is_receiver<>, is_single<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_receiver<>, is_single<>>>;
constexpr explicit single(Data d) constexpr explicit single(Data d)
: single(std::move(d), DVF{}, DEF{}, DDF{}) {} : single(std::move(d), DVF{}, DEF{}, DDF{}) {}
......
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// 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.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
...@@ -21,8 +23,10 @@ class any_single_sender { ...@@ -21,8 +23,10 @@ class any_single_sender {
} }
struct vtable { struct vtable {
static void s_op(data&, data*) {} static void s_op(data&, data*) {}
static any_time_executor<E /* hmm, TP will be invasive */> s_executor(data&) { return {}; }
static void s_submit(data&, single<V, E>) {} static void s_submit(data&, single<V, E>) {}
void (*op_)(data&, data*) = vtable::s_op; void (*op_)(data&, data*) = vtable::s_op;
any_time_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, single<V, E>) = vtable::s_submit; void (*submit_)(data&, single<V, E>) = vtable::s_submit;
}; };
static constexpr vtable const noop_ {}; static constexpr vtable const noop_ {};
...@@ -35,11 +39,14 @@ class any_single_sender { ...@@ -35,11 +39,14 @@ class any_single_sender {
dst->pobj_ = std::exchange(src.pobj_, nullptr); dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_); delete static_cast<Wrapped const*>(src.pobj_);
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, single<V, E> out) { static void submit(data& src, single<V, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out)); ::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
data_.pobj_ = new Wrapped(std::move(obj)); data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -53,12 +60,15 @@ class any_single_sender { ...@@ -53,12 +60,15 @@ class any_single_sender {
std::move(*static_cast<Wrapped*>((void*)src.buffer_))); std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped(); static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
} }
static any_time_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, single<V, E> out) { static void submit(data& src, single<V, E> out) {
::pushmi::submit( ::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out)); *static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
} }
}; };
static const vtable vtbl{s::op, s::submit}; static const vtable vtbl{s::op, s::executor, s::submit};
new (data_.buffer_) Wrapped(std::move(obj)); new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl; vptr_ = &vtbl;
} }
...@@ -87,6 +97,9 @@ class any_single_sender { ...@@ -87,6 +97,9 @@ class any_single_sender {
new ((void*)this) any_single_sender(std::move(that)); new ((void*)this) any_single_sender(std::move(that));
return *this; return *this;
} }
any_time_executor<E> executor() {
vptr_->executor_(data_);
}
void submit(single<V, E> out) { void submit(single<V, E> out) {
vptr_->submit_(data_, std::move(out)); vptr_->submit_(data_, std::move(out));
} }
...@@ -97,9 +110,10 @@ template <class V, class E> ...@@ -97,9 +110,10 @@ template <class V, class E>
constexpr typename any_single_sender<V, E>::vtable const constexpr typename any_single_sender<V, E>::vtable const
any_single_sender<V, E>::noop_; any_single_sender<V, E>::noop_;
template <class SF> template <class SF, class EXF>
class single_sender<SF> { class single_sender<SF, EXF> {
SF sf_; SF sf_;
EXF exf_;
public: public:
using properties = property_set<is_sender<>, is_single<>>; using properties = property_set<is_sender<>, is_single<>>;
...@@ -107,7 +121,10 @@ class single_sender<SF> { ...@@ -107,7 +121,10 @@ class single_sender<SF> {
constexpr single_sender() = default; constexpr single_sender() = default;
constexpr explicit single_sender(SF sf) constexpr explicit single_sender(SF sf)
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
constexpr single_sender(SF sf, EXF exf)
: sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>> PUSHMI_AND lazy::Invocable<SF&, Out>)) (requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>> PUSHMI_AND lazy::Invocable<SF&, Out>))
void submit(Out out) { void submit(Out out) {
...@@ -115,20 +132,24 @@ class single_sender<SF> { ...@@ -115,20 +132,24 @@ class single_sender<SF> {
} }
}; };
namespace detail { template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>>) Data, class DSF, class DEXF>
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>>) Data, class DSF> class single_sender<Data, DSF, DEXF> {
class single_sender_2 {
Data data_; Data data_;
DSF sf_; DSF sf_;
DEXF exf_;
public: public:
using properties = property_set<is_sender<>, is_single<>>; using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_single<>>>;
constexpr single_sender_2() = default; constexpr single_sender() = default;
constexpr explicit single_sender_2(Data data) constexpr explicit single_sender(Data data)
: data_(std::move(data)) {} : data_(std::move(data)) {}
constexpr single_sender_2(Data data, DSF sf) constexpr single_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {} : data_(std::move(data)), sf_(std::move(sf)) {}
constexpr single_sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>> PUSHMI_AND (requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>)) lazy::Invocable<DSF&, Data&, Out>))
...@@ -137,60 +158,63 @@ class single_sender_2 { ...@@ -137,60 +158,63 @@ class single_sender_2 {
} }
}; };
template <class A, class B>
using single_sender_base =
std::conditional_t<
(bool)Sender<A, is_single<>>,
single_sender_2<A, B>,
any_single_sender<A, B>>;
} // namespace detail
template <class A, class B>
struct single_sender<A, B>
: detail::single_sender_base<A, B> {
constexpr single_sender() = default;
using detail::single_sender_base<A, B>::single_sender_base;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_single_sender // make_single_sender
PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn {
inline auto operator()() const { inline auto operator()() const {
return single_sender<ignoreSF>{}; return single_sender<ignoreSF, trampolineEXF>{};
} }
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const { auto operator()(SF sf) const {
return single_sender<SF>{std::move(sf)}; return single_sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return single_sender<SF, EXF>{std::move(sf), std::move(exf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>>) (requires True<> && Sender<Data, is_single<>>)
auto operator()(Data d) const { auto operator()(Data d) const {
return single_sender<Data, passDSF>{std::move(d)}; return single_sender<Data, passDSF, passDEXF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>>) (requires Sender<Data, is_single<>>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return single_sender<Data, DSF>{std::move(d), std::move(sf)}; return single_sender<Data, DSF, passDEXF>{std::move(d), std::move(sf)};
}
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_single<>> && Invocable<DEXF&, Data&>)
auto operator()(Data d, DSF sf, DEXF exf) const {
return single_sender<Data, DSF, DEXF>{std::move(d), std::move(sf), std::move(exf)};
} }
} const make_single_sender {}; } const make_single_sender {};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// deduction guides // deduction guides
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
single_sender() -> single_sender<ignoreSF>; single_sender() -> single_sender<ignoreSF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF) PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>)) (requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
single_sender(SF) -> single_sender<SF>; single_sender(SF) -> single_sender<SF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
single_sender(SF, EXF) -> single_sender<SF, EXF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>>) (requires True<> && Sender<Data, is_single<>>)
single_sender(Data) -> single_sender<Data, passDSF>; single_sender(Data) -> single_sender<Data, passDSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>>) (requires Sender<Data, is_single<>>)
single_sender(Data, DSF) -> single_sender<Data, DSF>; single_sender(Data, DSF) -> single_sender<Data, DSF, passDEXF>;
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_single<>> && Invocable<DEXF&, Data&>)
single_sender(Data, DSF, DEXF) -> single_sender<Data, DSF, DEXF>;
#endif #endif
template<> template<>
......
...@@ -7,7 +7,8 @@ ...@@ -7,7 +7,8 @@
#include <vector> #include <vector>
#include <pushmi/time_single_sender.h> #include "time_single_sender.h"
#include "trampoline.h"
namespace pushmi { namespace pushmi {
...@@ -17,7 +18,7 @@ struct subject; ...@@ -17,7 +18,7 @@ struct subject;
template<class T, class PS> template<class T, class PS>
struct subject<T, PS> { struct subject<T, PS> {
using properties = property_set_insert_t<property_set<is_sender<>, is_single<>>, PS>; using properties = property_set_insert_t<property_set<is_sender<>, is_single<>>, property_set<property_set_index_t<PS, is_silent<>>>>;
struct subject_shared { struct subject_shared {
bool done_ = false; bool done_ = false;
...@@ -61,7 +62,7 @@ struct subject<T, PS> { ...@@ -61,7 +62,7 @@ struct subject<T, PS> {
// need a template overload of none/sender and the rest that stores a 'ptr' with its own lifetime management // need a template overload of none/sender and the rest that stores a 'ptr' with its own lifetime management
struct subject_receiver { struct subject_receiver {
using properties = property_set_insert_t<property_set<is_receiver<>, is_single<>>, PS>; using properties = property_set_insert_t<property_set<is_receiver<>, is_single<>>, property_set<property_set_index_t<PS, is_silent<>>>>;
std::shared_ptr<subject_shared> s; std::shared_ptr<subject_shared> s;
...@@ -82,6 +83,7 @@ struct subject<T, PS> { ...@@ -82,6 +83,7 @@ struct subject<T, PS> {
std::shared_ptr<subject_shared> s = std::make_shared<subject_shared>(); std::shared_ptr<subject_shared> s = std::make_shared<subject_shared>();
auto executor() { return trampoline(); }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) (requires Receiver<Out>)
void submit(Out out) { void submit(Out out) {
......
...@@ -9,7 +9,6 @@ ...@@ -9,7 +9,6 @@
#include <deque> #include <deque>
#include <thread> #include <thread>
#include "executor.h" #include "executor.h"
#include "time_single_sender.h"
namespace pushmi { namespace pushmi {
...@@ -41,12 +40,12 @@ class delegator : _pipeable_sender_ { ...@@ -41,12 +40,12 @@ class delegator : _pipeable_sender_ {
using time_point = typename trampoline<E>::time_point; using time_point = typename trampoline<E>::time_point;
public: public:
using properties = property_set<is_time<>, is_single<>>; using properties = property_set<is_time<>, is_executor<>, is_single<>>;
time_point now() { time_point now() {
return trampoline<E>::now(); return trampoline<E>::now();
} }
delegator executor() { return {}; }
PUSHMI_TEMPLATE (class SingleReceiver) PUSHMI_TEMPLATE (class SingleReceiver)
(requires Receiver<remove_cvref_t<SingleReceiver>, is_single<>>) (requires Receiver<remove_cvref_t<SingleReceiver>, is_single<>>)
void submit(time_point when, SingleReceiver&& what) { void submit(time_point when, SingleReceiver&& what) {
...@@ -60,12 +59,12 @@ class nester : _pipeable_sender_ { ...@@ -60,12 +59,12 @@ class nester : _pipeable_sender_ {
using time_point = typename trampoline<E>::time_point; using time_point = typename trampoline<E>::time_point;
public: public:
using properties = property_set<is_time<>, is_single<>>; using properties = property_set<is_time<>, is_executor<>, is_single<>>;
time_point now() { time_point now() {
return trampoline<E>::now(); return trampoline<E>::now();
} }
nester executor() { return {}; }
template <class SingleReceiver> template <class SingleReceiver>
void submit(time_point when, SingleReceiver&& what) { void submit(time_point when, SingleReceiver&& what) {
trampoline<E>::submit(ownornest, when, std::forward<SingleReceiver>(what)); trampoline<E>::submit(ownornest, when, std::forward<SingleReceiver>(what));
...@@ -257,6 +256,11 @@ inline detail::nester<E> nested_trampoline() { ...@@ -257,6 +256,11 @@ inline detail::nester<E> nested_trampoline() {
return {}; return {};
} }
// see boosters.h
struct trampolineEXF {
auto operator()() { return trampoline(); }
};
namespace detail { namespace detail {
PUSHMI_TEMPLATE (class E) PUSHMI_TEMPLATE (class E)
......
...@@ -6,7 +6,7 @@ target_include_directories(catch PUBLIC ...@@ -6,7 +6,7 @@ target_include_directories(catch PUBLIC
include(../external/Catch2/contrib/Catch.cmake) include(../external/Catch2/contrib/Catch.cmake)
if(NOT ${CMAKE_VERSION} VERSION_LESS "3.10.0") if(NOT ${CMAKE_VERSION} VERSION_LESS "3.10.0" AND NOT PUSHMI_ONE_TEST_BINARY)
add_executable(FlowTest FlowTest.cpp) add_executable(FlowTest FlowTest.cpp)
target_link_libraries(FlowTest pushmi catch Threads::Threads) target_link_libraries(FlowTest pushmi catch Threads::Threads)
......
...@@ -58,11 +58,12 @@ void none_test() { ...@@ -58,11 +58,12 @@ void none_test() {
void sender_test(){ void sender_test(){
auto in0 = pushmi::MAKE(sender)(); auto in0 = pushmi::MAKE(sender)();
auto in1 = pushmi::MAKE(sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(sender)(pushmi::ignoreSF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(sender)([&](auto out){ auto in3 = pushmi::MAKE(sender)([&](auto out){
in0.submit(pushmi::MAKE(none)(std::move(out), [](auto d, auto e) noexcept { in0.submit(pushmi::MAKE(none)(std::move(out), [](auto d, auto e) noexcept {
pushmi::set_error(d, e); pushmi::set_error(d, e);
})); }));
}); }, [](){ return pushmi::trampoline(); });
in3.submit(pushmi::MAKE(none)()); in3.submit(pushmi::MAKE(none)());
std::promise<void> p0; std::promise<void> p0;
...@@ -75,6 +76,8 @@ void sender_test(){ ...@@ -75,6 +76,8 @@ void sender_test(){
in3.submit(out1); in3.submit(out1);
auto any0 = pushmi::any_sender<>(in0); auto any0 = pushmi::any_sender<>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
void single_test() { void single_test() {
...@@ -192,11 +195,12 @@ void many_test() { ...@@ -192,11 +195,12 @@ void many_test() {
void single_sender_test(){ void single_sender_test(){
auto in0 = pushmi::MAKE(single_sender)(); auto in0 = pushmi::MAKE(single_sender)();
auto in1 = pushmi::MAKE(single_sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(single_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(single_sender)(pushmi::ignoreSF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(single_sender)([&](auto out){ auto in3 = pushmi::MAKE(single_sender)([&](auto out){
in0.submit(pushmi::MAKE(single)(std::move(out), in0.submit(pushmi::MAKE(single)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); }) pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
)); ));
}); }, [](){ return pushmi::trampoline(); });
std::promise<int> p0; std::promise<int> p0;
auto promise0 = pushmi::MAKE(single)(std::move(p0)); auto promise0 = pushmi::MAKE(single)(std::move(p0));
...@@ -209,16 +213,19 @@ void single_sender_test(){ ...@@ -209,16 +213,19 @@ void single_sender_test(){
in3.submit(out1); in3.submit(out1);
auto any0 = pushmi::any_single_sender<int>(in0); auto any0 = pushmi::any_single_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
void many_sender_test(){ void many_sender_test(){
auto in0 = pushmi::MAKE(many_sender)(); auto in0 = pushmi::MAKE(many_sender)();
auto in1 = pushmi::MAKE(many_sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(many_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(many_sender)(pushmi::ignoreSF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(many_sender)([&](auto out){ auto in3 = pushmi::MAKE(many_sender)([&](auto out){
in0.submit(pushmi::MAKE(many)(std::move(out), in0.submit(pushmi::MAKE(many)(std::move(out),
pushmi::on_next([](auto d, int v){ pushmi::set_next(d, v); }) pushmi::on_next([](auto d, int v){ pushmi::set_next(d, v); })
)); ));
}); }, [](){ return pushmi::trampoline(); });
auto out0 = pushmi::MAKE(many)(); auto out0 = pushmi::MAKE(many)();
auto out1 = pushmi::MAKE(many)(out0, pushmi::on_next([](auto d, int v){ auto out1 = pushmi::MAKE(many)(out0, pushmi::on_next([](auto d, int v){
...@@ -227,16 +234,19 @@ void many_sender_test(){ ...@@ -227,16 +234,19 @@ void many_sender_test(){
in3.submit(out1); in3.submit(out1);
auto any0 = pushmi::any_many_sender<int>(in0); auto any0 = pushmi::any_many_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
void time_single_sender_test(){ void time_single_sender_test(){
auto in0 = pushmi::MAKE(time_single_sender)(); auto in0 = pushmi::MAKE(time_single_sender)();
auto in1 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{}, pushmi::systemNowF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(time_single_sender)([&](auto tp, auto out){ auto in3 = pushmi::MAKE(time_single_sender)([&](auto tp, auto out){
in0.submit(tp, pushmi::MAKE(single)(std::move(out), in0.submit(tp, pushmi::MAKE(single)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); }) pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
)); ));
}); }, [](){ return pushmi::trampoline(); }, [](){ return std::chrono::system_clock::now(); });
auto in4 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{}, pushmi::systemNowF{}); auto in4 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{}, pushmi::systemNowF{});
std::promise<int> p0; std::promise<int> p0;
...@@ -251,6 +261,8 @@ void time_single_sender_test(){ ...@@ -251,6 +261,8 @@ void time_single_sender_test(){
auto any0 = pushmi::any_time_single_sender<int>(in0); auto any0 = pushmi::any_time_single_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
in3 | op::submit(); in3 | op::submit();
in3 | op::submit_at(in3.now() + 1s); in3 | op::submit_at(in3.now() + 1s);
in3 | op::submit_after(1s); in3 | op::submit_after(1s);
...@@ -368,11 +380,12 @@ void flow_single_test() { ...@@ -368,11 +380,12 @@ void flow_single_test() {
void flow_single_sender_test(){ void flow_single_sender_test(){
auto in0 = pushmi::MAKE(flow_single_sender)(); auto in0 = pushmi::MAKE(flow_single_sender)();
auto in1 = pushmi::MAKE(flow_single_sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(flow_single_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(flow_single_sender)(pushmi::ignoreSF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(flow_single_sender)([&](auto out){ auto in3 = pushmi::MAKE(flow_single_sender)([&](auto out){
in0.submit(pushmi::MAKE(flow_single)(std::move(out), in0.submit(pushmi::MAKE(flow_single)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); }) pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
)); ));
}); }, [](){ return pushmi::trampoline(); });
auto out0 = pushmi::MAKE(flow_single)(); auto out0 = pushmi::MAKE(flow_single)();
auto out1 = pushmi::MAKE(flow_single)(out0, pushmi::on_value([](auto d, int v){ auto out1 = pushmi::MAKE(flow_single)(out0, pushmi::on_value([](auto d, int v){
...@@ -381,6 +394,8 @@ void flow_single_sender_test(){ ...@@ -381,6 +394,8 @@ void flow_single_sender_test(){
in3.submit(out1); in3.submit(out1);
auto any0 = pushmi::any_flow_single_sender<int>(in0); auto any0 = pushmi::any_flow_single_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
void flow_many_test() { void flow_many_test() {
...@@ -453,11 +468,12 @@ void flow_many_test() { ...@@ -453,11 +468,12 @@ void flow_many_test() {
void flow_many_sender_test(){ void flow_many_sender_test(){
auto in0 = pushmi::MAKE(flow_many_sender)(); auto in0 = pushmi::MAKE(flow_many_sender)();
auto in1 = pushmi::MAKE(flow_many_sender)(pushmi::ignoreSF{}); auto in1 = pushmi::MAKE(flow_many_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(flow_many_sender)(pushmi::ignoreSF{}, pushmi::trampolineEXF{});
auto in3 = pushmi::MAKE(flow_many_sender)([&](auto out){ auto in3 = pushmi::MAKE(flow_many_sender)([&](auto out){
in0.submit(pushmi::MAKE(flow_many)(std::move(out), in0.submit(pushmi::MAKE(flow_many)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); }) pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
)); ));
}); }, [](){ return pushmi::trampoline(); });
auto out0 = pushmi::MAKE(flow_many)(); auto out0 = pushmi::MAKE(flow_many)();
auto out1 = pushmi::MAKE(flow_many)(out0, pushmi::on_value([](auto d, int v){ auto out1 = pushmi::MAKE(flow_many)(out0, pushmi::on_value([](auto d, int v){
...@@ -466,4 +482,6 @@ void flow_many_sender_test(){ ...@@ -466,4 +482,6 @@ void flow_many_sender_test(){
in3.submit(out1); in3.submit(out1);
auto any0 = pushmi::any_flow_many_sender<int>(in0); auto any0 = pushmi::any_flow_many_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
...@@ -15,8 +15,8 @@ using namespace std::literals; ...@@ -15,8 +15,8 @@ using namespace std::literals;
#include "pushmi/o/submit.h" #include "pushmi/o/submit.h"
#include "pushmi/o/extension_operators.h" #include "pushmi/o/extension_operators.h"
#include "pushmi/inline.h"
#include "pushmi/trampoline.h" #include "pushmi/trampoline.h"
#include "pushmi/new_thread.h"
using namespace pushmi::aliases; using namespace pushmi::aliases;
...@@ -142,11 +142,15 @@ SCENARIO( "trampoline executor", "[trampoline][sender]" ) { ...@@ -142,11 +142,15 @@ SCENARIO( "trampoline executor", "[trampoline][sender]" ) {
::pushmi::set_value(out, std::numeric_limits<int8_t>::min()); ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::pushmi::set_value(out, std::numeric_limits<int8_t>::max()); ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
}); });
sender | op::on([&](){return tr;}) | auto inlineon = sender | op::on([&](){return mi::inline_executor();});
inlineon |
op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); })); op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); }));
THEN( "only the first item was pushed" ) { THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"}); REQUIRE(values == std::vector<std::string>{"2.000000"});
} }
THEN( "executor was not changed by on" ) {
REQUIRE(std::is_same<mi::executor_t<decltype(sender)>, mi::executor_t<decltype(inlineon)>>::value);
}
} }
WHEN( "used with via" ) { WHEN( "used with via" ) {
...@@ -158,11 +162,15 @@ SCENARIO( "trampoline executor", "[trampoline][sender]" ) { ...@@ -158,11 +162,15 @@ SCENARIO( "trampoline executor", "[trampoline][sender]" ) {
::pushmi::set_value(out, std::numeric_limits<int8_t>::min()); ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::pushmi::set_value(out, std::numeric_limits<int8_t>::max()); ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
}); });
sender | op::via([&](){return tr;}) | auto inlinevia = sender | op::via([&](){return mi::inline_executor();});
inlinevia |
op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); })); op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); }));
THEN( "only the first item was pushed" ) { THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"}); REQUIRE(values == std::vector<std::string>{"2.000000"});
} }
THEN( "executor was changed by via" ) {
REQUIRE(!std::is_same<mi::executor_t<decltype(sender)>, mi::executor_t<decltype(inlinevia)>>::value);
}
} }
} }
} }
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