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

refactoring time senders and executors (#51)

fbshipit-source-id: 6df047bf7833fe622b6c9901912b8022e485a610
parent 5160a6b0
FIND_PACKAGE (Boost) FIND_PACKAGE (Boost)
if (Boost_FOUND) if (Boost_FOUND)
...@@ -13,7 +14,6 @@ target_include_directories(PushmiBenchmarks ...@@ -13,7 +14,6 @@ target_include_directories(PushmiBenchmarks
target_link_libraries(PushmiBenchmarks target_link_libraries(PushmiBenchmarks
pushmi pushmi
Threads::Threads Threads::Threads
${Boost_LIBRARIES}
) )
else() else()
......
...@@ -12,6 +12,7 @@ ...@@ -12,6 +12,7 @@
#include "pushmi/trampoline.h" #include "pushmi/trampoline.h"
#include "pushmi/new_thread.h" #include "pushmi/new_thread.h"
#include "pushmi/time_source.h"
#include "pushmi/none.h" #include "pushmi/none.h"
#include "pushmi/entangle.h" #include "pushmi/entangle.h"
...@@ -82,7 +83,7 @@ struct countdownnone { ...@@ -82,7 +83,7 @@ struct countdownnone {
struct inline_time_executor { struct inline_time_executor {
using properties = mi::property_set<mi::is_time<>, mi::is_executor<>, 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 top() { return std::chrono::system_clock::now(); }
auto executor() { return *this; } 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) {
...@@ -437,7 +438,7 @@ NONIUS_BENCHMARK("trampoline virtual derecursion 1'000", [](nonius::chronometer ...@@ -437,7 +438,7 @@ NONIUS_BENCHMARK("trampoline virtual derecursion 1'000", [](nonius::chronometer
auto tr = mi::trampoline(); auto tr = mi::trampoline();
using TR = decltype(tr); using TR = decltype(tr);
auto single = countdownsingle{counter}; auto single = countdownsingle{counter};
std::function<void(mi::any_time_executor_ref<>)> recurse{[&](auto exec){::pushmi::set_value(single, exec);}}; std::function<void(mi::any_executor_ref<>)> recurse{[&](auto exec){::pushmi::set_value(single, exec);}};
meter.measure([&]{ meter.measure([&]{
counter.store(1'000); counter.store(1'000);
tr | op::submit([&](auto exec) { recurse(exec); }); tr | op::submit([&](auto exec) { recurse(exec); });
...@@ -503,3 +504,32 @@ NONIUS_BENCHMARK("new thread submit 1'000", [](nonius::chronometer meter){ ...@@ -503,3 +504,32 @@ NONIUS_BENCHMARK("new thread submit 1'000", [](nonius::chronometer meter){
return counter.load(); return counter.load();
}); });
}) })
NONIUS_BENCHMARK("new thread blocking_submit 1'000", [](nonius::chronometer meter){
auto nt = mi::new_thread();
using NT = decltype(nt);
std::atomic<int> counter{0};
countdownsingle single{counter};
meter.measure([&]{
counter.store(1'000);
nt | op::blocking_submit(single);
return counter.load();
});
})
NONIUS_BENCHMARK("new thread + time submit 1'000", [](nonius::chronometer meter){
auto nt = mi::new_thread();
using NT = decltype(nt);
auto time = mi::time_source<>{};
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })();
using TNT = decltype(tnt);
std::atomic<int> counter{0};
countdownsingle single{counter};
meter.measure([&]{
counter.store(1'000);
tnt | op::submit(single);
while(counter.load() > 0);
return counter.load();
});
time.join();
})
...@@ -42,7 +42,9 @@ set(header_files ...@@ -42,7 +42,9 @@ set(header_files
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/inline.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/constrained_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_source.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/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/single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
......
...@@ -24,21 +24,17 @@ using std::experimental::static_thread_pool; ...@@ -24,21 +24,17 @@ using std::experimental::static_thread_pool;
namespace execution = std::experimental::execution; namespace execution = std::experimental::execution;
template<class Executor> template<class Executor>
struct pool_time_executor { struct pool_executor {
using properties = property_set<is_time<>, is_executor<>, is_single<>>; using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
using e_t = Executor; using e_t = Executor;
e_t e; e_t e;
explicit pool_time_executor(e_t e) : e(std::move(e)) {} explicit pool_executor(e_t e) : e(std::move(e)) {}
auto now() {
return std::chrono::system_clock::now();
}
auto executor() { return *this; } auto executor() { return *this; }
PUSHMI_TEMPLATE(class TP, class Out) PUSHMI_TEMPLATE(class Out)
(requires Regular<TP> && Receiver<Out>) (requires Receiver<Out>)
void submit(TP at, Out out) const { void submit(Out out) const {
e.execute([e = *this, at = std::move(at), out = std::move(out)]() mutable { e.execute([e = *this, out = std::move(out)]() mutable {
std::this_thread::sleep_until(at);
::pushmi::set_value(out, e); ::pushmi::set_value(out, e);
}); });
} }
...@@ -52,7 +48,7 @@ public: ...@@ -52,7 +48,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 pool_time_executor<decltype(exec)>{exec}; return pool_executor<decltype(exec)>{exec};
} }
inline void stop() {p.stop();} inline void stop() {p.stop();}
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <tuple> #include <tuple>
#include <deque> #include <deque>
#include <vector> #include <vector>
#include <queue>
#if __cpp_lib_optional >= 201606 #if __cpp_lib_optional >= 201606
#include <optional> #include <optional>
......
This source diff could not be displayed because it is too large. You can view the blob instead.
...@@ -56,6 +56,9 @@ struct construct_deduced<many_sender>; ...@@ -56,6 +56,9 @@ struct construct_deduced<many_sender>;
template<> template<>
struct construct_deduced<flow_single_sender>; struct construct_deduced<flow_single_sender>;
template<>
struct construct_deduced<constrained_single_sender>;
template<> template<>
struct construct_deduced<time_single_sender>; struct construct_deduced<time_single_sender>;
...@@ -93,7 +96,6 @@ struct trampolineEXF; ...@@ -93,7 +96,6 @@ struct trampolineEXF;
// auto operator()() { return trampoline(); } // auto operator()() { return trampoline(); }
// }; // };
struct ignoreSF { struct ignoreSF {
void operator()(detail::any) {} void operator()(detail::any) {}
void operator()(detail::any, detail::any) {} void operator()(detail::any, detail::any) {}
...@@ -103,6 +105,10 @@ struct systemNowF { ...@@ -103,6 +105,10 @@ struct systemNowF {
auto operator()() { return std::chrono::system_clock::now(); } auto operator()() { return std::chrono::system_clock::now(); }
}; };
struct priorityZeroF {
auto operator()(){ return 0; }
};
struct passDVF { struct passDVF {
PUSHMI_TEMPLATE(class V, class Data) PUSHMI_TEMPLATE(class V, class Data)
(requires requires ( (requires requires (
...@@ -176,6 +182,14 @@ struct passDNF { ...@@ -176,6 +182,14 @@ struct passDNF {
} }
}; };
struct passDZF {
PUSHMI_TEMPLATE(class Data)
(requires ConstrainedSender<Data>)
auto operator()(Data& in) const noexcept {
return ::pushmi::top(in);
}
};
// inspired by Ovrld - shown in a presentation by Nicolai Josuttis // inspired by Ovrld - shown in a presentation by Nicolai Josuttis
#if __cpp_variadic_using >= 201611 && __cpp_concepts #if __cpp_variadic_using >= 201611 && __cpp_concepts
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... Fns> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... Fns>
......
...@@ -170,38 +170,38 @@ PUSHMI_CONCEPT_DEF( ...@@ -170,38 +170,38 @@ PUSHMI_CONCEPT_DEF(
is_executor_v<PS> && is_sender_v<PS> && is_single_v<PS> is_executor_v<PS> && is_sender_v<PS> && is_single_v<PS>
); );
// Time trait and tag // Constrained trait and tag
template<class... TN> template<class... TN>
struct is_time; struct is_constrained;
// Tag // Tag
template<> template<>
struct is_time<> : is_sender<> {}; struct is_constrained<> : is_sender<> {};
// Trait // Trait
template<class PS> template<class PS>
struct is_time<PS> : property_query<PS, is_time<>> {}; struct is_constrained<PS> : property_query<PS, is_constrained<>> {};
template<class PS> template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_time_v = is_time<PS>::value; 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 Time, concept Constrained,
is_time_v<PS> && is_sender_v<PS> is_constrained_v<PS> && is_sender_v<PS>
); );
// Constrained trait and tag // Time trait and tag
template<class... TN> template<class... TN>
struct is_constrained; struct is_time;
// Tag // Tag
template<> template<>
struct is_constrained<> : is_sender<> {}; struct is_time<> : is_constrained<> {};
// Trait // Trait
template<class PS> template<class PS>
struct is_constrained<PS> : property_query<PS, is_constrained<>> {}; struct is_time<PS> : property_query<PS, is_time<>> {};
template<class PS> template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_constrained_v = is_constrained<PS>::value; 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 Constrained, concept Time,
is_constrained_v<PS> && is_sender_v<PS> is_time_v<PS> && is_constrained_v<PS> && is_sender_v<PS>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
...@@ -358,70 +358,64 @@ PUSHMI_CONCEPT_DEF( ...@@ -358,70 +358,64 @@ PUSHMI_CONCEPT_DEF(
// add concepts for constraints // add concepts for constraints
// //
// the constraint could be time or priority enum or any other
// ordering constraint value-type.
//
// top() returns the constraint value that will cause the item to run asap.
// So now() for time and NORMAL for priority.
//
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class D, class... PropertyN) template (class D, class... PropertyN)
(concept TimeSender)(D, PropertyN...), (concept ConstrainedSender)(D, PropertyN...),
requires(D& d) ( requires(D& d) (
::pushmi::now(d), ::pushmi::top(d),
requires_<Regular<decltype(::pushmi::now(d))>> requires_<Regular<decltype(::pushmi::top(d))>>
) && ) &&
Sender<D> && Sender<D> &&
property_query_v<D, PropertyN...> && property_query_v<D, PropertyN...> &&
Time<D> && Constrained<D> &&
None<D> None<D>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class D, class S, class... PropertyN) template (class D, class S, class... PropertyN)
(concept TimeSenderTo)(D, S, PropertyN...), (concept ConstrainedSenderTo)(D, S, PropertyN...),
requires(D& d, S&& s) ( requires(D& d, S&& s) (
::pushmi::submit(d, ::pushmi::now(d), (S &&) s) ::pushmi::submit(d, ::pushmi::top(d), (S &&) s)
) && ) &&
TimeSender<D> && ConstrainedSender<D> &&
property_query_v<D, PropertyN...> && property_query_v<D, PropertyN...> &&
Receiver<S> Receiver<S>
); );
template <class D> template <class D>
PUSHMI_PP_CONSTRAINED_USING( PUSHMI_PP_CONSTRAINED_USING(
TimeSender<D>, ConstrainedSender<D>,
time_point_t =, decltype(::pushmi::now(std::declval<D&>()))); constraint_t =, decltype(::pushmi::top(std::declval<D&>())));
// this is a more general form where the constraint could be time or priority
// enum or any other ordering constraint value-type.
//
// top() returns the constraint value that will cause the item to run asap.
// So now() for time and NORMAL for priority.
//
// I would like to replace Time.. with Constrained.. but not sure if it will
// obscure too much.
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class D) template (class D, class... PropertyN)
concept ConstrainedSender, (concept TimeSender)(D, PropertyN...),
requires(D& d) ( requires(D& d) (
::pushmi::top(d), ::pushmi::now(d),
requires_<Regular<decltype(::pushmi::top(d))>> requires_<Regular<decltype(::pushmi::now(d) + std::chrono::seconds(1))>>
) && ) &&
Sender<D> && ConstrainedSender<D, PropertyN...> &&
Constrained<D> && Time<D>
None<D>
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class D, class S) template (class D, class S, class... PropertyN)
concept ConstrainedSenderTo, (concept TimeSenderTo)(D, S, PropertyN...),
requires(D& d, S&& s) ( ConstrainedSenderTo<D, S, PropertyN...> &&
::pushmi::submit(d, ::pushmi::top(d), (S &&) s) TimeSender<D>
) &&
ConstrainedSender<D> &&
Receiver<S>
); );
template <class D> template <class D>
PUSHMI_PP_CONSTRAINED_USING( PUSHMI_PP_CONSTRAINED_USING(
ConstrainedSender<D>, TimeSender<D>,
constraint_t =, decltype(::pushmi::top(std::declval<D&>()))); time_point_t =, decltype(::pushmi::now(std::declval<D&>())));
} // namespace pushmi } // namespace pushmi
...@@ -56,15 +56,15 @@ void submit(SD& sd, Out out) noexcept(noexcept(sd.submit(std::move(out)))) { ...@@ -56,15 +56,15 @@ void submit(SD& sd, Out out) noexcept(noexcept(sd.submit(std::move(out)))) {
} }
PUSHMI_TEMPLATE (class SD) PUSHMI_TEMPLATE (class SD)
(requires requires (std::declval<SD&>().now())) (requires requires (std::declval<SD&>().top()))
auto now(SD& sd) noexcept(noexcept(sd.now())) { auto top(SD& sd) noexcept(noexcept(sd.top())) {
return sd.now(); return sd.top();
} }
PUSHMI_TEMPLATE (class SD, class TP, class Out) PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires ( (requires requires (
std::declval<SD&>().submit( std::declval<SD&>().submit(
std::declval<TP(&)(TP)>()(std::declval<SD&>().now()), std::declval<TP(&)(TP)>()(std::declval<SD&>().top()),
std::declval<Out>()) std::declval<Out>())
)) ))
void submit(SD& sd, TP tp, Out out) void submit(SD& sd, TP tp, Out out)
...@@ -116,15 +116,15 @@ void submit(SD& sd, Out out) noexcept(noexcept(sd->submit(std::move(out)))) { ...@@ -116,15 +116,15 @@ void submit(SD& sd, Out out) noexcept(noexcept(sd->submit(std::move(out)))) {
} }
PUSHMI_TEMPLATE (class SD) PUSHMI_TEMPLATE (class SD)
(requires requires (std::declval<SD&>()->now())) (requires requires (std::declval<SD&>()->top()))
auto now(SD& sd) noexcept(noexcept(sd->now())) { auto top(SD& sd) noexcept(noexcept(sd->top())) {
return sd->now(); return sd->top();
} }
PUSHMI_TEMPLATE (class SD, class TP, class Out) PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires ( (requires requires (
std::declval<SD&>()->submit( std::declval<SD&>()->submit(
std::declval<TP(&)(TP)>()(std::declval<SD&>()->now()), std::declval<TP(&)(TP)>()(std::declval<SD&>()->top()),
std::declval<Out>()) std::declval<Out>())
)) ))
void submit(SD& sd, TP tp, Out out) void submit(SD& sd, TP tp, Out out)
...@@ -203,15 +203,15 @@ void submit(std::reference_wrapper<SD> sd, Out out) noexcept( ...@@ -203,15 +203,15 @@ void submit(std::reference_wrapper<SD> sd, Out out) noexcept(
submit(sd.get(), std::move(out)); submit(sd.get(), std::move(out));
} }
PUSHMI_TEMPLATE (class SD) PUSHMI_TEMPLATE (class SD)
(requires requires ( now(std::declval<SD&>()) )) (requires requires ( top(std::declval<SD&>()) ))
auto now(std::reference_wrapper<SD> sd) noexcept(noexcept(now(sd.get()))) { auto top(std::reference_wrapper<SD> sd) noexcept(noexcept(top(sd.get()))) {
return now(sd.get()); return top(sd.get());
} }
PUSHMI_TEMPLATE (class SD, class TP, class Out) PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires ( (requires requires (
submit( submit(
std::declval<SD&>(), std::declval<SD&>(),
std::declval<TP(&)(TP)>()(now(std::declval<SD&>())), std::declval<TP(&)(TP)>()(top(std::declval<SD&>())),
std::declval<Out>()) std::declval<Out>())
)) ))
void submit(std::reference_wrapper<SD> sd, TP tp, Out out) void submit(std::reference_wrapper<SD> sd, TP tp, Out out)
...@@ -327,13 +327,13 @@ struct do_submit_fn { ...@@ -327,13 +327,13 @@ struct do_submit_fn {
} }
}; };
struct get_now_fn { struct get_top_fn {
PUSHMI_TEMPLATE (class SD) PUSHMI_TEMPLATE (class SD)
(requires requires ( (requires requires (
now(std::declval<SD&>()) top(std::declval<SD&>())
)) ))
auto operator()(SD&& sd) const noexcept(noexcept(now(sd))) { auto operator()(SD&& sd) const noexcept(noexcept(top(sd))) {
return now(sd); return top(sd);
} }
}; };
...@@ -346,8 +346,8 @@ PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{}; ...@@ -346,8 +346,8 @@ 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::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_top_fn now{};
PUSHMI_INLINE_VAR constexpr __adl::get_now_fn top{}; PUSHMI_INLINE_VAR constexpr __adl::get_top_fn top{};
template <class T> template <class T>
struct property_set_traits<std::promise<T>> { struct property_set_traits<std::promise<T>> {
......
...@@ -23,10 +23,10 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -23,10 +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 any_executor<E> 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; any_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_ {};
...@@ -39,8 +39,8 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -39,8 +39,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))}; return any_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));
...@@ -60,8 +60,8 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -60,8 +60,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))}; return any_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(
...@@ -97,7 +97,7 @@ class flow_many_sender<V, PV, PE, E> { ...@@ -97,7 +97,7 @@ 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() { any_executor<E> executor() {
return vptr_->executor_(data_); return vptr_->executor_(data_);
} }
void submit(any_flow_many<V, PV, PE, E> out) { void submit(any_flow_many<V, PV, PE, E> out) {
...@@ -158,6 +158,13 @@ class flow_many_sender<Data, DSF, DEXF> { ...@@ -158,6 +158,13 @@ class flow_many_sender<Data, DSF, DEXF> {
} }
}; };
template <>
class flow_many_sender<>
: public flow_many_sender<ignoreSF, trampolineEXF> {
public:
flow_many_sender() = default;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// 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 {
......
...@@ -23,10 +23,10 @@ class any_flow_single_sender { ...@@ -23,10 +23,10 @@ class any_flow_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 any_executor<E> 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; any_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_ {};
...@@ -39,8 +39,8 @@ class any_flow_single_sender { ...@@ -39,8 +39,8 @@ class any_flow_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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))}; return any_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));
...@@ -60,8 +60,8 @@ class any_flow_single_sender { ...@@ -60,8 +60,8 @@ class any_flow_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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))}; return any_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(
...@@ -97,7 +97,7 @@ class any_flow_single_sender { ...@@ -97,7 +97,7 @@ class any_flow_single_sender {
new ((void*)this) any_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() { any_executor<E> executor() {
return vptr_->executor_(data_); return vptr_->executor_(data_);
} }
void submit(flow_single<V, PE, E> out) { void submit(flow_single<V, PE, E> out) {
...@@ -158,6 +158,13 @@ class flow_single_sender<Data, DSF, DEXF> { ...@@ -158,6 +158,13 @@ class flow_single_sender<Data, DSF, DEXF> {
} }
}; };
template <>
class flow_single_sender<>
: public flow_single_sender<ignoreSF, trampolineEXF> {
public:
flow_single_sender() = default;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// 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 {
......
...@@ -65,6 +65,9 @@ class single_sender; ...@@ -65,6 +65,9 @@ class single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class many_sender; class many_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class constrained_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class time_single_sender; class time_single_sender;
...@@ -80,11 +83,41 @@ class flow_many; ...@@ -80,11 +83,41 @@ 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 any_single_sender;
template<
class V,
class E = std::exception_ptr,
class C = std::ptrdiff_t>
struct any_constrained_single_sender;
template< template<
class V, class V,
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>
struct any_time_single_sender; class any_time_single_sender;
template<
class E = std::exception_ptr>
struct any_executor;
template<
class E = std::exception_ptr>
struct any_executor_ref;
template<
class E = std::exception_ptr,
class CV = std::ptrdiff_t>
struct any_constrained_executor;
template<
class E = std::exception_ptr,
class TP = std::ptrdiff_t>
struct any_constrained_executor_ref;
template< template<
class E = std::exception_ptr, class E = std::exception_ptr,
......
...@@ -8,11 +8,36 @@ ...@@ -8,11 +8,36 @@
namespace pushmi { namespace pushmi {
class inline_time_executor { class inline_constrained_executor_t {
public:
using properties = property_set<is_constrained<>, is_executor<>, is_single<>>;
std::ptrdiff_t top() {
return 0;
}
auto executor() { return *this; }
PUSHMI_TEMPLATE(class CV, class Out)
(requires Regular<CV> && Receiver<Out, is_single<>>)
void submit(CV, Out out) {
::pushmi::set_value(std::move(out), *this);
}
};
struct inlineConstrainedEXF {
inline_constrained_executor_t operator()(){
return {};
}
};
inline inline_constrained_executor_t inline_constrained_executor() {
return {};
}
class inline_time_executor_t {
public: public:
using properties = property_set<is_time<>, is_executor<>, is_single<>>; using properties = property_set<is_time<>, is_executor<>, is_single<>>;
auto now() { auto top() {
return std::chrono::system_clock::now(); return std::chrono::system_clock::now();
} }
auto executor() { return *this; } auto executor() { return *this; }
...@@ -24,13 +49,35 @@ class inline_time_executor { ...@@ -24,13 +49,35 @@ class inline_time_executor {
} }
}; };
struct inlineTimeEXF {
inline_time_executor_t operator()(){
return {};
}
};
inline inline_time_executor_t inline_time_executor() {
return {};
}
class inline_executor_t {
public:
using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
auto executor() { return *this; }
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_single<>>)
void submit(Out out) {
::pushmi::set_value(std::move(out), *this);
}
};
struct inlineEXF { struct inlineEXF {
inline_time_executor operator()(){ inline_executor_t operator()(){
return {}; return {};
} }
}; };
inline inline_time_executor inline_executor() { inline inline_executor_t inline_executor() {
return {}; return {};
} }
......
...@@ -23,10 +23,10 @@ class any_many_sender { ...@@ -23,10 +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 any_executor<E> 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; any_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_ {};
...@@ -39,8 +39,8 @@ class any_many_sender { ...@@ -39,8 +39,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))}; return any_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));
...@@ -60,8 +60,8 @@ class any_many_sender { ...@@ -60,8 +60,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))}; return any_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(
...@@ -97,7 +97,7 @@ class any_many_sender { ...@@ -97,7 +97,7 @@ 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() { any_executor<E> executor() {
return vptr_->executor_(data_); return vptr_->executor_(data_);
} }
void submit(many<V, E> out) { void submit(many<V, E> out) {
...@@ -158,6 +158,13 @@ class many_sender<Data, DSF, DEXF> { ...@@ -158,6 +158,13 @@ class many_sender<Data, DSF, DEXF> {
} }
}; };
template <>
class many_sender<>
: public many_sender<ignoreSF, trampolineEXF> {
public:
many_sender() = default;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_many_sender // make_many_sender
PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn {
......
...@@ -12,26 +12,23 @@ namespace pushmi { ...@@ -12,26 +12,23 @@ namespace pushmi {
// very poor perf example executor. // very poor perf example executor.
// //
struct new_thread_time_executor { struct new_thread_executor {
using properties = property_set<is_time<>, is_executor<>, is_single<>>; using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
auto now() { new_thread_executor executor() { return {}; }
return std::chrono::system_clock::now(); PUSHMI_TEMPLATE(class Out)
} (requires Receiver<Out>)
new_thread_time_executor executor() { return {}; } void submit(Out out) {
PUSHMI_TEMPLATE(class TP, class Out) std::thread t{[out = std::move(out)]() mutable {
(requires Regular<TP> && Receiver<Out>)
void submit(TP at, Out out) {
std::thread t{[at = std::move(at), out = std::move(out)]() mutable {
auto tr = ::pushmi::trampoline(); auto tr = ::pushmi::trampoline();
::pushmi::submit(tr, std::move(at), std::move(out)); ::pushmi::submit(tr, std::move(out));
}}; }};
// pass ownership of thread to out // pass ownership of thread to out
t.detach(); t.detach();
} }
}; };
inline new_thread_time_executor new_thread() { inline new_thread_executor new_thread() {
return {}; return {};
} }
......
...@@ -155,7 +155,7 @@ auto submit_transform_out(SDSF, TSDSF tsdsf) { ...@@ -155,7 +155,7 @@ auto submit_transform_out(SDSF, TSDSF tsdsf) {
return on_submit(submit_transform_out_4<In, TSDSF>{std::move(tsdsf)}); return on_submit(submit_transform_out_4<In, TSDSF>{std::move(tsdsf)});
} }
template <class Cardinality, bool IsTime = false, bool IsFlow = false> template <class Cardinality, bool IsConstrained = false, bool IsTime = false, bool IsFlow = false>
struct make_sender; struct make_sender;
template <> template <>
struct make_sender<is_none<>> : construct_deduced<sender> {}; struct make_sender<is_none<>> : construct_deduced<sender> {};
...@@ -164,11 +164,13 @@ struct make_sender<is_single<>> : construct_deduced<single_sender> {}; ...@@ -164,11 +164,13 @@ struct make_sender<is_single<>> : construct_deduced<single_sender> {};
template <> template <>
struct make_sender<is_many<>> : construct_deduced<many_sender> {}; struct make_sender<is_many<>> : construct_deduced<many_sender> {};
template <> template <>
struct make_sender<is_single<>, false, true> : construct_deduced<flow_single_sender> {}; struct make_sender<is_single<>, false, false, true> : construct_deduced<flow_single_sender> {};
template <> template <>
struct make_sender<is_many<>, false, true> : construct_deduced<flow_many_sender> {}; struct make_sender<is_many<>, false, false, true> : construct_deduced<flow_many_sender> {};
template <> template <>
struct make_sender<is_single<>, true, false> : construct_deduced<time_single_sender> {}; struct make_sender<is_single<>, true, true, false> : construct_deduced<time_single_sender> {};
template <>
struct make_sender<is_single<>, true, false, false> : construct_deduced<constrained_single_sender> {};
PUSHMI_INLINE_VAR constexpr struct sender_from_fn { PUSHMI_INLINE_VAR constexpr struct sender_from_fn {
PUSHMI_TEMPLATE(class In, class... FN) PUSHMI_TEMPLATE(class In, class... FN)
...@@ -177,6 +179,7 @@ PUSHMI_INLINE_VAR constexpr struct sender_from_fn { ...@@ -177,6 +179,7 @@ PUSHMI_INLINE_VAR constexpr struct sender_from_fn {
using MakeSender = using MakeSender =
make_sender< make_sender<
property_set_index_t<properties_t<In>, is_silent<>>, property_set_index_t<properties_t<In>, is_silent<>>,
property_query_v<properties_t<In>, is_constrained<>>,
property_query_v<properties_t<In>, is_time<>>, property_query_v<properties_t<In>, is_time<>>,
property_query_v<properties_t<In>, is_flow<>>>; property_query_v<properties_t<In>, is_flow<>>>;
return MakeSender{}(std::move(in), (FN&&) fn...); return MakeSender{}(std::move(in), (FN&&) fn...);
...@@ -299,7 +302,7 @@ private: ...@@ -299,7 +302,7 @@ private:
struct impl { struct impl {
PUSHMI_TEMPLATE (class In) PUSHMI_TEMPLATE (class In)
(requires Sender<In>) (requires Sender<In>)
auto operator()(In in) const { auto operator()(In& in) const {
return ::pushmi::executor(in); return ::pushmi::executor(in);
} }
}; };
...@@ -316,7 +319,7 @@ private: ...@@ -316,7 +319,7 @@ private:
Out out_; Out out_;
PUSHMI_TEMPLATE (class In) PUSHMI_TEMPLATE (class In)
(requires SenderTo<In, Out>) (requires SenderTo<In, Out>)
void operator()(In in) { void operator()(In& in) {
::pushmi::submit(in, std::move(out_)); ::pushmi::submit(in, std::move(out_));
} }
}; };
...@@ -326,7 +329,7 @@ private: ...@@ -326,7 +329,7 @@ private:
Out out_; Out out_;
PUSHMI_TEMPLATE (class In) PUSHMI_TEMPLATE (class In)
(requires TimeSenderTo<In, Out>) (requires TimeSenderTo<In, Out>)
void operator()(In in) { void operator()(In& in) {
::pushmi::submit(in, std::move(tp_), std::move(out_)); ::pushmi::submit(in, std::move(tp_), std::move(out_));
} }
}; };
...@@ -343,12 +346,27 @@ public: ...@@ -343,12 +346,27 @@ public:
} }
}; };
struct top_fn {
private:
struct impl {
PUSHMI_TEMPLATE (class In)
(requires ConstrainedSender<In>)
auto operator()(In& in) const {
return ::pushmi::top(in);
}
};
public:
auto operator()() const {
return impl{};
}
};
struct now_fn { struct now_fn {
private: private:
struct impl { struct impl {
PUSHMI_TEMPLATE (class In) PUSHMI_TEMPLATE (class In)
(requires TimeSender<In>) (requires TimeSender<In>)
auto operator()(In in) const { auto operator()(In& in) const {
return ::pushmi::now(in); return ::pushmi::now(in);
} }
}; };
...@@ -370,7 +388,7 @@ PUSHMI_INLINE_VAR constexpr detail::set_starting_fn set_starting{}; ...@@ -370,7 +388,7 @@ PUSHMI_INLINE_VAR constexpr detail::set_starting_fn set_starting{};
PUSHMI_INLINE_VAR constexpr detail::executor_fn executor{}; 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::top_fn top{};
} // namespace extension_operators } // namespace extension_operators
......
...@@ -81,7 +81,6 @@ struct flow_from_up { ...@@ -81,7 +81,6 @@ struct flow_from_up {
if (requested < 1) {return;} if (requested < 1) {return;}
// submit work to exec // submit work to exec
::pushmi::submit(p->exec, ::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p, requested](auto) { make_single([p = p, requested](auto) {
auto remaining = requested; auto remaining = requested;
// this loop is structured to work when there is re-entrancy // this loop is structured to work when there is re-entrancy
...@@ -102,7 +101,6 @@ struct flow_from_up { ...@@ -102,7 +101,6 @@ struct flow_from_up {
void error(E e) noexcept { void error(E e) noexcept {
p->stop.store(true); p->stop.store(true);
::pushmi::submit(p->exec, ::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p](auto) { make_single([p = p](auto) {
::pushmi::set_done(p->out); ::pushmi::set_done(p->out);
})); }));
...@@ -111,7 +109,6 @@ struct flow_from_up { ...@@ -111,7 +109,6 @@ struct flow_from_up {
void done() { void done() {
p->stop.store(true); p->stop.store(true);
::pushmi::submit(p->exec, ::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p](auto) { make_single([p = p](auto) {
::pushmi::set_done(p->out); ::pushmi::set_done(p->out);
})); }));
...@@ -132,7 +129,6 @@ private: ...@@ -132,7 +129,6 @@ private:
auto p = std::make_shared<Producer>(begin_, end_, std::move(out), exec_, false); auto p = std::make_shared<Producer>(begin_, end_, std::move(out), exec_, false);
::pushmi::submit(exec_, ::pushmi::submit(exec_,
::pushmi::now(exec_),
make_single([p](auto exec) { make_single([p](auto exec) {
// pass reference for cancellation. // pass reference for cancellation.
::pushmi::set_starting(p->out, make_many(flow_from_up<Producer>{p})); ::pushmi::set_starting(p->out, make_many(flow_from_up<Producer>{p}));
...@@ -160,13 +156,13 @@ public: ...@@ -160,13 +156,13 @@ public:
DerivedFrom< DerivedFrom<
typename std::iterator_traits<I>::iterator_category, typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag> && std::forward_iterator_tag> &&
Sender<Exec> && Time<Exec> && Single<Exec>) Sender<Exec, is_single<>, is_executor<>>)
auto operator()(I begin, S end, Exec exec) const { auto operator()(I begin, S end, Exec exec) const {
return make_flow_many_sender(out_impl<I, S, Exec>{begin, end, exec}); return make_flow_many_sender(out_impl<I, S, Exec>{begin, end, exec});
} }
PUSHMI_TEMPLATE(class R, class Exec) PUSHMI_TEMPLATE(class R, class Exec)
(requires Range<R> && Sender<Exec> && Time<Exec> && Single<Exec>) (requires Range<R> && Sender<Exec, is_single<>, is_executor<>>)
auto operator()(R&& range, Exec exec) const { auto operator()(R&& range, Exec exec) const {
return (*this)(std::begin(range), std::end(range), exec); return (*this)(std::begin(range), std::end(range), exec);
} }
......
...@@ -29,7 +29,7 @@ private: ...@@ -29,7 +29,7 @@ private:
(requires SenderTo<In, Out>) (requires SenderTo<In, Out>)
void operator()(In& in, Out out) const { void operator()(In& in, Out out) const {
auto exec = ef_(); auto exec = ef_();
::pushmi::submit(exec, ::pushmi::now(exec), ::pushmi::submit(exec,
::pushmi::make_single(on_value_impl<In, Out>{in, std::move(out)}) ::pushmi::make_single(on_value_impl<In, Out>{in, std::move(out)})
); );
} }
...@@ -74,7 +74,7 @@ private: ...@@ -74,7 +74,7 @@ private:
}; };
public: public:
PUSHMI_TEMPLATE(class ExecutorFactory) PUSHMI_TEMPLATE(class ExecutorFactory)
(requires Invocable<ExecutorFactory&>) (requires Invocable<ExecutorFactory&> && Executor<invoke_result_t<ExecutorFactory&>>)
auto operator()(ExecutorFactory ef) const { auto operator()(ExecutorFactory ef) const {
return in_impl<ExecutorFactory>{std::move(ef)}; return in_impl<ExecutorFactory>{std::move(ef)};
} }
......
...@@ -30,6 +30,11 @@ PUSHMI_CONCEPT_DEF( ...@@ -30,6 +30,11 @@ PUSHMI_CONCEPT_DEF(
(concept AutoSenderTo)(In, AN...), (concept AutoSenderTo)(In, AN...),
SenderTo<In, receiver_type_t<In, AN...>> SenderTo<In, receiver_type_t<In, AN...>>
); );
PUSHMI_CONCEPT_DEF(
template (class In, class ... AN)
(concept AutoConstrainedSenderTo)(In, AN...),
ConstrainedSenderTo<In, receiver_type_t<In, AN...>> && not Time<In>
);
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class In, class ... AN) template (class In, class ... AN)
(concept AutoTimeSenderTo)(In, AN...), (concept AutoTimeSenderTo)(In, AN...),
...@@ -52,6 +57,13 @@ private: ...@@ -52,6 +57,13 @@ private:
::pushmi::submit(in, std::move(out)); ::pushmi::submit(in, std::move(out));
return in; return in;
} }
PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoConstrainedSenderTo<In, AN...>)
In operator()(In in) {
auto out{::pushmi::detail::receiver_from_fn<In>()(std::move(args_))};
::pushmi::submit(in, ::pushmi::top(in), std::move(out));
return in;
}
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoTimeSenderTo<In, AN...>) (requires submit_detail::AutoTimeSenderTo<In, AN...>)
In operator()(In in) { In operator()(In in) {
...@@ -118,28 +130,111 @@ public: ...@@ -118,28 +130,111 @@ public:
struct blocking_submit_fn { struct blocking_submit_fn {
private: private:
struct lock_state { struct lock_state {
bool done = false; bool done{false};
std::atomic<int> nested{0};
std::mutex lock; std::mutex lock;
std::condition_variable signaled; std::condition_variable signaled;
}; };
template<class Out>
struct nested_receiver_impl;
PUSHMI_TEMPLATE (class Exec)
(requires Sender<Exec> && Executor<Exec>)
struct nested_executor_impl {
nested_executor_impl(lock_state* state, Exec ex) :
state_(state),
ex_(std::move(ex)) {}
lock_state* state_;
Exec ex_;
using properties = properties_t<Exec>;
auto executor() { return ::pushmi::executor(ex_); }
PUSHMI_TEMPLATE (class CV, class Out)
(requires Receiver<Out> && Constrained<Exec>)
auto top() { return ::pushmi::top(ex_); }
PUSHMI_TEMPLATE (class CV, class Out)
(requires Receiver<Out> && Constrained<Exec>)
void submit(CV cv, Out out) {
++state_->nested;
::pushmi::submit(ex_, cv, nested_receiver_impl<Out>{state_, std::move(out)});
}
PUSHMI_TEMPLATE (class Out)
(requires Receiver<Out> && not Constrained<Exec>)
void submit(Out out) {
++state_->nested;
::pushmi::submit(ex_, nested_receiver_impl<Out>{state_, std::move(out)});
}
};
template<class Out>
struct nested_receiver_impl {
nested_receiver_impl(lock_state* state, Out out) :
state_(state),
out_(std::move(out)) {}
lock_state* state_;
Out out_;
using properties = properties_t<Out>;
template<class V>
void value(V&& v) {
std::exception_ptr e;
try{
using executor_t = remove_cvref_t<V>;
auto n = nested_executor_impl<executor_t>{state_, (V&&) v};
::pushmi::set_value(out_, any_executor_ref<>{n});
}
catch(...) {e = std::current_exception();}
if(--state_->nested == 0) {
state_->signaled.notify_all();
}
if (e) {std::rethrow_exception(e);}
}
template<class E>
void error(E&& e) noexcept {
::pushmi::set_error(out_, (E&&) e);
if(--state_->nested == 0) {
state_->signaled.notify_all();
}
}
void done() {
std::exception_ptr e;
try{
::pushmi::set_done(out_);
}
catch(...) {e = std::current_exception();}
if(--state_->nested == 0) {
state_->signaled.notify_all();
}
if (e) {std::rethrow_exception(e);}
}
};
struct nested_executor_impl_fn {
PUSHMI_TEMPLATE (class Exec)
(requires Executor<Exec>)
auto operator()(lock_state* state, Exec ex) const {
return nested_executor_impl<Exec>{state, std::move(ex)};
}
};
struct on_value_impl { struct on_value_impl {
lock_state* state_; lock_state* state_;
PUSHMI_TEMPLATE (class Out, class Value) PUSHMI_TEMPLATE (class Out, class Value)
(requires Receiver<Out, is_single<>>) (requires Receiver<Out, is_single<>>)
void operator()(Out out, Value&& v) const { void operator()(Out out, Value&& v) const {
using V = remove_cvref_t<Value>; using V = remove_cvref_t<Value>;
PUSHMI_IF_CONSTEXPR( ((bool)Time<V>) ( ++state_->nested;
// to keep the blocking semantics, make sure that the PUSHMI_IF_CONSTEXPR( ((bool)Executor<V>) (
// nested submits block here to prevent a spurious id(::pushmi::set_value)(out, id(nested_executor_impl_fn{})(state_, id((Value&&) v)));
// completion signal
auto nest = ::pushmi::nested_trampoline();
::pushmi::submit(nest, ::pushmi::now(nest), std::move(out));
) else ( ) else (
::pushmi::set_value(out, id((Value&&) v)); id(::pushmi::set_value)(out, id((Value&&) v));
)) ))
std::unique_lock<std::mutex> guard{state_->lock}; std::unique_lock<std::mutex> guard{state_->lock};
state_->done = true; state_->done = true;
state_->signaled.notify_all(); if (--state_->nested == 0){
state_->signaled.notify_all();
}
} }
}; };
struct on_next_impl { struct on_next_impl {
...@@ -172,7 +267,7 @@ private: ...@@ -172,7 +267,7 @@ private:
state_->signaled.notify_all(); state_->signaled.notify_all();
} }
}; };
template <bool IsTimeSender, class In> template <bool IsConstrainedSender, bool IsTimeSender, class In>
struct submit_impl { struct submit_impl {
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) (requires Receiver<Out>)
...@@ -180,7 +275,11 @@ private: ...@@ -180,7 +275,11 @@ private:
PUSHMI_IF_CONSTEXPR( (IsTimeSender) ( PUSHMI_IF_CONSTEXPR( (IsTimeSender) (
id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out)); id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out));
) else ( ) else (
id(::pushmi::submit)(in, std::move(out)); PUSHMI_IF_CONSTEXPR( (IsConstrainedSender) (
id(::pushmi::submit)(in, id(::pushmi::top)(in), std::move(out));
) else (
id(::pushmi::submit)(in, std::move(out));
))
)) ))
} }
}; };
...@@ -191,11 +290,11 @@ private: ...@@ -191,11 +290,11 @@ private:
struct fn { struct fn {
std::tuple<AN...> args_; std::tuple<AN...> args_;
template <bool IsTimeSender, class In> template <bool IsConstrainedSender, bool IsTimeSender, class In>
In impl_(In in) { In impl_(In in) {
lock_state state{}; lock_state state{};
auto submit = submit_impl<IsTimeSender, In>{}; auto submit = submit_impl<IsConstrainedSender, IsTimeSender, In>{};
PUSHMI_IF_CONSTEXPR( ((bool)Many<In>) ( PUSHMI_IF_CONSTEXPR( ((bool)Many<In>) (
auto out{::pushmi::detail::receiver_from_fn<In>()( auto out{::pushmi::detail::receiver_from_fn<In>()(
std::move(args_), std::move(args_),
...@@ -216,7 +315,7 @@ private: ...@@ -216,7 +315,7 @@ private:
std::unique_lock<std::mutex> guard{state.lock}; std::unique_lock<std::mutex> guard{state.lock};
state.signaled.wait(guard, [&]{ state.signaled.wait(guard, [&]{
return state.done; return state.done && state.nested.load() == 0;
}); });
return in; return in;
} }
...@@ -224,12 +323,17 @@ private: ...@@ -224,12 +323,17 @@ private:
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoSenderTo<In, AN...>) (requires submit_detail::AutoSenderTo<In, AN...>)
In operator()(In in) { In operator()(In in) {
return this->impl_<false>(std::move(in)); return this->impl_<false, false>(std::move(in));
}
PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoConstrainedSenderTo<In, AN...>)
In operator()(In in) {
return this->impl_<true, false>(std::move(in));
} }
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoTimeSenderTo<In, AN...>) (requires submit_detail::AutoTimeSenderTo<In, AN...>)
In operator()(In in) { In operator()(In in) {
return this->impl_<true>(std::move(in)); return this->impl_<true, true>(std::move(in));
} }
}; };
public: public:
......
...@@ -41,7 +41,6 @@ private: ...@@ -41,7 +41,6 @@ private:
void operator()(Data& data, V&& v) const { void operator()(Data& data, V&& v) const {
::pushmi::submit( ::pushmi::submit(
data.exec, data.exec,
::pushmi::now(data.exec),
::pushmi::make_single( ::pushmi::make_single(
impl<std::decay_t<V>>{(V&&) v, std::move(static_cast<Out&>(data))} impl<std::decay_t<V>>{(V&&) v, std::move(static_cast<Out&>(data))}
) )
...@@ -62,7 +61,6 @@ private: ...@@ -62,7 +61,6 @@ private:
void operator()(Data& data, E e) const noexcept { void operator()(Data& data, E e) const noexcept {
::pushmi::submit( ::pushmi::submit(
data.exec, data.exec,
::pushmi::now(data.exec),
::pushmi::make_single( ::pushmi::make_single(
impl<E>{std::move(e), std::move(static_cast<Out&>(data))} impl<E>{std::move(e), std::move(static_cast<Out&>(data))}
) )
...@@ -81,7 +79,6 @@ private: ...@@ -81,7 +79,6 @@ private:
void operator()(Data& data) const { void operator()(Data& data) const {
::pushmi::submit( ::pushmi::submit(
data.exec, data.exec,
::pushmi::now(data.exec),
::pushmi::make_single( ::pushmi::make_single(
impl{std::move(static_cast<Out&>(data))} impl{std::move(static_cast<Out&>(data))}
) )
...@@ -128,7 +125,7 @@ private: ...@@ -128,7 +125,7 @@ private:
}; };
public: public:
PUSHMI_TEMPLATE(class ExecutorFactory) PUSHMI_TEMPLATE(class ExecutorFactory)
(requires Invocable<ExecutorFactory&>) (requires Invocable<ExecutorFactory&> && Executor<invoke_result_t<ExecutorFactory&>>)
auto operator()(ExecutorFactory ef) const { auto operator()(ExecutorFactory ef) const {
return in_impl<ExecutorFactory>{std::move(ef)}; return in_impl<ExecutorFactory>{std::move(ef)};
} }
......
...@@ -26,10 +26,10 @@ class sender<detail::erase_sender_t, E> { ...@@ -26,10 +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 any_executor<E> 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; any_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_ {};
...@@ -42,8 +42,8 @@ class sender<detail::erase_sender_t, E> { ...@@ -42,8 +42,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))}; return any_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));
...@@ -62,8 +62,8 @@ class sender<detail::erase_sender_t, E> { ...@@ -62,8 +62,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))}; return any_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(
...@@ -97,7 +97,7 @@ class sender<detail::erase_sender_t, E> { ...@@ -97,7 +97,7 @@ 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() { any_executor<E> executor() {
return vptr_->executor_(data_); return vptr_->executor_(data_);
} }
void submit(any_none<E> out) { void submit(any_none<E> out) {
...@@ -157,6 +157,13 @@ class sender<Data, DSF, DEXF> { ...@@ -157,6 +157,13 @@ class sender<Data, DSF, DEXF> {
} }
}; };
template <>
class sender<>
: public sender<ignoreSF, trampolineEXF> {
public:
sender() = default;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_sender // make_sender
PUSHMI_INLINE_VAR constexpr struct make_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_sender_fn {
......
...@@ -10,7 +10,7 @@ ...@@ -10,7 +10,7 @@
namespace pushmi { namespace pushmi {
template <class V, class E = std::exception_ptr> template <class V, class E>
class any_single_sender { class any_single_sender {
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
...@@ -23,10 +23,10 @@ class any_single_sender { ...@@ -23,10 +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 any_executor<E> 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; any_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_ {};
...@@ -39,8 +39,8 @@ class any_single_sender { ...@@ -39,8 +39,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))}; return any_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));
...@@ -60,8 +60,8 @@ class any_single_sender { ...@@ -60,8 +60,8 @@ 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) { static any_executor<E> executor(data& src) {
return any_time_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))}; return any_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(
...@@ -97,7 +97,7 @@ class any_single_sender { ...@@ -97,7 +97,7 @@ 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() { any_executor<E> executor() {
vptr_->executor_(data_); vptr_->executor_(data_);
} }
void submit(single<V, E> out) { void submit(single<V, E> out) {
...@@ -158,6 +158,13 @@ class single_sender<Data, DSF, DEXF> { ...@@ -158,6 +158,13 @@ class single_sender<Data, DSF, DEXF> {
} }
}; };
template <>
class single_sender<>
: public single_sender<ignoreSF, trampolineEXF> {
public:
single_sender() = default;
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_single_sender // make_single_sender
PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn { PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn {
......
This diff is collapsed.
...@@ -37,51 +37,38 @@ class trampoline; ...@@ -37,51 +37,38 @@ class trampoline;
template <class E = std::exception_ptr> template <class E = std::exception_ptr>
class delegator : _pipeable_sender_ { class delegator : _pipeable_sender_ {
using time_point = typename trampoline<E>::time_point;
public: public:
using properties = property_set<is_time<>, is_executor<>, is_single<>>; using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
time_point now() {
return trampoline<E>::now();
}
delegator executor() { return {}; } 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(SingleReceiver&& what) {
trampoline<E>::submit( trampoline<E>::submit(
ownordelegate, when, std::forward<SingleReceiver>(what)); ownordelegate, std::forward<SingleReceiver>(what));
} }
}; };
template <class E = std::exception_ptr> template <class E = std::exception_ptr>
class nester : _pipeable_sender_ { class nester : _pipeable_sender_ {
using time_point = typename trampoline<E>::time_point;
public: public:
using properties = property_set<is_time<>, is_executor<>, is_single<>>; using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
time_point now() {
return trampoline<E>::now();
}
nester executor() { return {}; } nester executor() { return {}; }
template <class SingleReceiver> template <class SingleReceiver>
void submit(time_point when, SingleReceiver&& what) { void submit(SingleReceiver&& what) {
trampoline<E>::submit(ownornest, when, std::forward<SingleReceiver>(what)); trampoline<E>::submit(ownornest, std::forward<SingleReceiver>(what));
} }
}; };
template <class E> template <class E>
class trampoline { class trampoline {
public:
using time_point = std::chrono::system_clock::time_point;
private: private:
using error_type = std::decay_t<E>; using error_type = std::decay_t<E>;
using work_type = using work_type =
any_single<any_time_executor_ref<error_type, time_point>, error_type>; any_single<any_executor_ref<error_type>, error_type>;
using queue_type = std::deque<std::tuple<time_point, work_type>>; using queue_type = std::deque<work_type>;
using pending_type = std::tuple<int, queue_type, time_point>; using pending_type = std::tuple<int, queue_type, bool>;
inline static pending_type*& owner() { inline static pending_type*& owner() {
static thread_local pending_type* pending = nullptr; static thread_local pending_type* pending = nullptr;
...@@ -96,7 +83,7 @@ class trampoline { ...@@ -96,7 +83,7 @@ class trampoline {
return std::get<1>(p); return std::get<1>(p);
} }
inline static time_point& next(pending_type& p) { inline static bool& repeat(pending_type& p) {
return std::get<2>(p); return std::get<2>(p);
} }
...@@ -109,21 +96,17 @@ class trampoline { ...@@ -109,21 +96,17 @@ class trampoline {
return owner() != nullptr; return owner() != nullptr;
} }
inline static time_point now() {
return std::chrono::system_clock::now();
}
template <class Selector, class Derived> template <class Selector, class Derived>
static void submit(Selector, Derived&, time_point awhen, recurse_t) { static void submit(Selector, Derived&, recurse_t) {
if (!is_owned()) { if (!is_owned()) {
abort(); abort();
} }
next(*owner()) = awhen; repeat(*owner()) = true;
} }
PUSHMI_TEMPLATE (class SingleReceiver) PUSHMI_TEMPLATE (class SingleReceiver)
(requires not Same<SingleReceiver, recurse_t>) (requires not Same<SingleReceiver, recurse_t>)
static void submit(ownordelegate_t, time_point awhen, SingleReceiver awhat) { static void submit(ownordelegate_t, SingleReceiver awhat) {
delegator<E> that; delegator<E> that;
if (is_owned()) { if (is_owned()) {
...@@ -131,10 +114,9 @@ class trampoline { ...@@ -131,10 +114,9 @@ class trampoline {
// poor mans scope guard // poor mans scope guard
try { try {
if (++depth(*owner()) > 100 || awhen > trampoline<E>::now()) { if (++depth(*owner()) > 100) {
// defer work to owner // defer work to owner
pending(*owner()).push_back( pending(*owner()).push_back(work_type{std::move(awhat)});
std::make_tuple(awhen, work_type{std::move(awhat)}));
} else { } else {
// dynamic recursion - optimization to balance queueing and // dynamic recursion - optimization to balance queueing and
// stack usage and value interleaving on the same thread. // stack usage and value interleaving on the same thread.
...@@ -153,16 +135,16 @@ class trampoline { ...@@ -153,16 +135,16 @@ class trampoline {
pending_type pending_store; pending_type pending_store;
owner() = &pending_store; owner() = &pending_store;
depth(pending_store) = 0; depth(pending_store) = 0;
repeat(pending_store) = false;
// poor mans scope guard // poor mans scope guard
try { try {
trampoline<E>::submit(ownornest, awhen, std::move(awhat)); trampoline<E>::submit(ownornest, std::move(awhat));
} catch(...) { } catch(...) {
// ignore exceptions while delivering the exception // ignore exceptions while delivering the exception
try { try {
::pushmi::set_error(awhat, std::current_exception()); ::pushmi::set_error(awhat, std::current_exception());
for (auto& item : pending(pending_store)) { for (auto& what : pending(pending_store)) {
auto& what = std::get<1>(item);
::pushmi::set_error(what, std::current_exception()); ::pushmi::set_error(what, std::current_exception());
} }
} catch (...) { } catch (...) {
...@@ -181,11 +163,11 @@ class trampoline { ...@@ -181,11 +163,11 @@ class trampoline {
PUSHMI_TEMPLATE (class SingleReceiver) PUSHMI_TEMPLATE (class SingleReceiver)
(requires not Same<SingleReceiver, recurse_t>) (requires not Same<SingleReceiver, recurse_t>)
static void submit(ownornest_t, time_point awhen, SingleReceiver awhat) { static void submit(ownornest_t, SingleReceiver awhat) {
delegator<E> that; delegator<E> that;
if (!is_owned()) { if (!is_owned()) {
trampoline<E>::submit(ownordelegate, awhen, std::move(awhat)); trampoline<E>::submit(ownordelegate, std::move(awhat));
return; return;
} }
...@@ -193,19 +175,15 @@ class trampoline { ...@@ -193,19 +175,15 @@ class trampoline {
// static recursion - tail call optimization // static recursion - tail call optimization
if (pending(pending_store).empty()) { if (pending(pending_store).empty()) {
auto when = awhen; bool go = true;
while (when != time_point{}) { while (go) {
if (when > trampoline<E>::now()) { repeat(pending_store) = false;
std::this_thread::sleep_until(when);
}
next(pending_store) = time_point{};
::pushmi::set_value(awhat, that); ::pushmi::set_value(awhat, that);
when = next(pending_store); go = repeat(pending_store);
} }
} else { } else {
// ensure work is sorted by time
pending(pending_store) pending(pending_store)
.push_back(std::make_tuple(awhen, work_type{std::move(awhat)})); .push_back(work_type{std::move(awhat)});
} }
if (pending(pending_store).empty()) { if (pending(pending_store).empty()) {
...@@ -213,22 +191,9 @@ class trampoline { ...@@ -213,22 +191,9 @@ class trampoline {
} }
while (!pending(pending_store).empty()) { while (!pending(pending_store).empty()) {
std::stable_sort( auto what = std::move(pending(pending_store).front());
pending(pending_store).begin(),
pending(pending_store).end(),
[](auto& lhs, auto& rhs) {
auto& lwhen = std::get<0>(lhs);
auto& rwhen = std::get<0>(rhs);
return lwhen < rwhen;
});
auto item = std::move(pending(pending_store).front());
pending(pending_store).pop_front(); pending(pending_store).pop_front();
auto& when = std::get<0>(item); any_executor_ref<error_type> anythis{that};
if (when > trampoline<E>::now()) {
std::this_thread::sleep_until(when);
}
auto& what = std::get<1>(item);
any_time_executor_ref<error_type, time_point> anythis{that};
::pushmi::set_value(what, anythis); ::pushmi::set_value(what, anythis);
} }
} }
...@@ -264,9 +229,9 @@ struct trampolineEXF { ...@@ -264,9 +229,9 @@ struct trampolineEXF {
namespace detail { namespace detail {
PUSHMI_TEMPLATE (class E) PUSHMI_TEMPLATE (class E)
(requires TimeSenderTo<delegator<E>, recurse_t>) (requires SenderTo<delegator<E>, recurse_t>)
decltype(auto) repeat(delegator<E>& exec) { decltype(auto) repeat(delegator<E>& exec) {
::pushmi::submit(exec, ::pushmi::now(exec), recurse); ::pushmi::submit(exec, recurse);
} }
template <class AnyExec> template <class AnyExec>
void repeat(AnyExec& exec) { void repeat(AnyExec& exec) {
......
...@@ -39,7 +39,8 @@ add_executable(PushmiTest PushmiTest.cpp ...@@ -39,7 +39,8 @@ add_executable(PushmiTest PushmiTest.cpp
TrampolineTest.cpp TrampolineTest.cpp
NewThreadTest.cpp NewThreadTest.cpp
FlowTest.cpp FlowTest.cpp
FlowManyTest.cpp) FlowManyTest.cpp
)
target_link_libraries(PushmiTest pushmi catch Threads::Threads) target_link_libraries(PushmiTest pushmi catch Threads::Threads)
catch_discover_tests(PushmiTest) catch_discover_tests(PushmiTest)
......
...@@ -238,76 +238,64 @@ void many_sender_test(){ ...@@ -238,76 +238,64 @@ void many_sender_test(){
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor"); static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
} }
void constrained_single_sender_test(){
auto in0 = pushmi::MAKE(constrained_single_sender)();
auto in1 = pushmi::MAKE(constrained_single_sender)(pushmi::ignoreSF{});
auto in2 = pushmi::MAKE(constrained_single_sender)(pushmi::ignoreSF{}, pushmi::inlineConstrainedEXF{}, pushmi::priorityZeroF{});
auto in3 = pushmi::MAKE(constrained_single_sender)([&](auto c, auto out){
in0.submit(c, pushmi::MAKE(single)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
));
}, [](){ return pushmi::inline_constrained_executor(); }, [](){ return 0; });
auto in4 = pushmi::MAKE(constrained_single_sender)(pushmi::ignoreSF{}, pushmi::inlineConstrainedEXF{});
std::promise<int> p0;
auto promise0 = pushmi::MAKE(single)(std::move(p0));
in0.submit(in0.top(), std::move(promise0));
auto out0 = pushmi::MAKE(single)();
auto out1 = pushmi::MAKE(single)(out0, pushmi::on_value([](auto d, int v){
pushmi::set_value(d, v);
}));
in3.submit(in0.top(), out1);
auto any0 = pushmi::any_constrained_single_sender<int>(in0);
static_assert(pushmi::Executor<pushmi::executor_t<decltype(in0)>>, "sender has invalid executor");
in3 | op::submit();
in3 | op::blocking_submit();
}
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 in2 = pushmi::MAKE(time_single_sender)(pushmi::ignoreSF{}, pushmi::inlineTimeEXF{}, pushmi::systemNowF{});
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(); }); }, [](){ return pushmi::inline_time_executor(); }, [](){ 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::inlineTimeEXF{});
std::promise<int> p0; std::promise<int> p0;
auto promise0 = pushmi::MAKE(single)(std::move(p0)); auto promise0 = pushmi::MAKE(single)(std::move(p0));
in0.submit(in0.now(), std::move(promise0)); in0.submit(in0.top(), std::move(promise0));
auto out0 = pushmi::MAKE(single)(); auto out0 = pushmi::MAKE(single)();
auto out1 = pushmi::MAKE(single)(out0, pushmi::on_value([](auto d, int v){ auto out1 = pushmi::MAKE(single)(out0, pushmi::on_value([](auto d, int v){
pushmi::set_value(d, v); pushmi::set_value(d, v);
})); }));
in3.submit(in0.now(), out1); in3.submit(in0.top(), out1);
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"); 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::blocking_submit();
in3 | op::submit_at(in3.top() + 1s);
in3 | op::submit_after(1s); in3 | op::submit_after(1s);
#if 0
auto tr = v::trampoline();
tr |
op::transform([]<class TR>(TR tr) {
return v::get_now(tr);
}) |
// op::submit(v::MAKE(single){});
op::get<std::chrono::system_clock::time_point>();
std::vector<std::string> times;
auto push = [&](int time) {
return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); });
};
auto nt = v::new_thread();
auto out = v::any_single<v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>>{v::MAKE(single){}};
(v::any_time_executor_ref{nt}).submit(v::get_now(nt), v::MAKE(single){});
nt |
op::transform([&]<class NT>(NT nt){
// auto out = v::any_single<v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>>{v::MAKE(single){}};
// nt.submit(v::get_now(nt), std::move(out));
// nt.submit(v::get_now(nt), v::MAKE(single){});
// nt | op::submit_at(v::get_now(nt), std::move(out));
// nt |
// op::submit(v::MAKE(single){});
// op::submit_at(nt | ep::get_now(), v::on_value{[](auto){}}) |
// op::submit_at(nt | ep::get_now(), v::on_value{[](auto){}}) |
// op::submit_after(20ms, v::MAKE(single){}) ;//|
// op::submit_after(20ms, v::on_value{[](auto){}}) |
// op::submit_after(40ms, push(42));
return v::get_now(nt);
}) |
// op::submit(v::MAKE(single){});
op::blocking_submit(v::MAKE(single){});
// op::get<decltype(v::get_now(nt))>();
// op::get<std::chrono::system_clock::time_point>();
auto ref = v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>{nt};
#endif
} }
void flow_single_test() { void flow_single_test() {
......
...@@ -14,6 +14,7 @@ using namespace std::literals; ...@@ -14,6 +14,7 @@ using namespace std::literals;
#include "pushmi/entangle.h" #include "pushmi/entangle.h"
#include "pushmi/new_thread.h" #include "pushmi/new_thread.h"
#include "pushmi/time_source.h"
#include "pushmi/trampoline.h" #include "pushmi/trampoline.h"
using namespace pushmi::aliases; using namespace pushmi::aliases;
...@@ -26,7 +27,7 @@ using namespace pushmi::aliases; ...@@ -26,7 +27,7 @@ using namespace pushmi::aliases;
#define MAKE(x) make_##x #define MAKE(x) make_##x
#endif #endif
SCENARIO("flow many immediate cancellation", "[flow][sender]") { SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") {
int signals = 0; int signals = 0;
GIVEN("A flow many sender") { GIVEN("A flow many sender") {
...@@ -104,109 +105,15 @@ SCENARIO("flow many immediate cancellation", "[flow][sender]") { ...@@ -104,109 +105,15 @@ SCENARIO("flow many immediate cancellation", "[flow][sender]") {
} }
} }
SCENARIO("flow many cancellation trampoline", "[flow][sender]") { SCENARIO("flow many cancellation new thread", "[flowmany][flow][sender]") {
auto tr = mi::trampoline();
using TR = decltype(tr);
int signals = 0;
GIVEN("A flow many sender") {
auto f = mi::MAKE(flow_many_sender)([&](auto out) {
using Out = decltype(out);
struct Data : mi::many<> {
explicit Data(Out out) : out(std::move(out)), stop(false), valid(true) {}
Out out;
bool stop;
bool valid;
};
auto up = mi::MAKE(many)(
Data{std::move(out)},
[&](auto& data, auto requested) {
signals += 1000000;
if (requested < 1 || data.stop) {return;}
// submit work to happen later
auto datal = std::move(data);
data.valid = false;
tr |
op::submit_after(
100ms,
[data = std::move(datal)](auto) mutable {
// check boolean to select signal
if (!data.stop) {
::mi::set_next(data.out, 42);
}
::mi::set_done(data.out);
});
},
[&](auto& data, auto e) noexcept {
signals += 100000;
if(data.stop || !data.valid) {return;}
data.stop = true;
::mi::set_done(data.out);
},
[&](auto& data) {
signals += 10000;
if(data.stop || !data.valid) {return;}
data.stop = true;
::mi::set_done(data.out);
});
tr |
op::submit([up = std::move(up)](auto tr) mutable {
// pass reference for cancellation.
::mi::set_starting(up.data().out, std::move(up));
});
});
WHEN("submit is applied and cancels the producer") {
f |
op::submit(
mi::on_next([&](int) { signals += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }),
mi::on_done([&]() { signals += 1; }),
// stop producer before it is scheduled to run
mi::on_starting([&](auto up) {
signals += 10;
tr | op::submit_after(50ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
THEN(
"the starting, up.done and out.done signals are each recorded once") {
REQUIRE(signals == 10011);
}
}
WHEN("submit is applied and cancels the producer late") {
f |
op::submit(
mi::on_next([&](int) { signals += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }),
mi::on_done([&]() { signals += 1; }),
// do not stop producer before it is scheduled to run
mi::on_starting([&](auto up) {
signals += 10;
mi::set_next(up, 1);
tr |
op::submit_after(250ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
THEN(
"the starting, up.done and out.value signals are each recorded once") {
REQUIRE(signals == 1010111);
}
}
}
}
SCENARIO("flow many cancellation new thread", "[flow][sender]") {
auto nt = mi::new_thread(); auto nt = mi::new_thread();
using NT = decltype(nt); using NT = decltype(nt);
auto time = mi::time_source<>{};
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })();
using TNT = decltype(tnt);
auto tcncl = time.make(mi::systemNowF{}, [nt](){ return nt; })();
std::atomic<int> signals{0}; std::atomic<int> signals{0};
auto at = nt.now() + 200ms; auto at = mi::now(tnt) + 200ms;
GIVEN("A flow many sender") { GIVEN("A flow many sender") {
auto f = mi::MAKE(flow_many_sender)([&](auto out) { auto f = mi::MAKE(flow_many_sender)([&](auto out) {
...@@ -214,12 +121,12 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -214,12 +121,12 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
// boolean cancellation // boolean cancellation
struct producer { struct producer {
producer(Out out, NT nt, bool s) : out(std::move(out)), nt(std::move(nt)), stop(s) {} producer(Out out, TNT tnt, bool s) : out(std::move(out)), tnt(std::move(tnt)), stop(s) {}
Out out; Out out;
NT nt; TNT tnt;
std::atomic<bool> stop; std::atomic<bool> stop;
}; };
auto p = std::make_shared<producer>(std::move(out), nt, false); auto p = std::make_shared<producer>(std::move(out), tnt, false);
struct Data : mi::many<> { struct Data : mi::many<> {
explicit Data(std::shared_ptr<producer> p) : p(std::move(p)) {} explicit Data(std::shared_ptr<producer> p) : p(std::move(p)) {}
...@@ -232,7 +139,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -232,7 +139,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
signals += 1000000; signals += 1000000;
if (requested < 1) {return;} if (requested < 1) {return;}
// submit work to happen later // submit work to happen later
data.p->nt | data.p->tnt |
op::submit_at( op::submit_at(
at, at,
[p = data.p](auto) { [p = data.p](auto) {
...@@ -246,20 +153,20 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -246,20 +153,20 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
[&signals](auto& data, auto e) noexcept { [&signals](auto& data, auto e) noexcept {
signals += 100000; signals += 100000;
data.p->stop.store(true); data.p->stop.store(true);
data.p->nt | op::submit([p = data.p](auto) { data.p->tnt | op::submit([p = data.p](auto) {
::mi::set_done(p->out); ::mi::set_done(p->out);
}); });
}, },
[&signals](auto& data) { [&signals](auto& data) {
signals += 10000; signals += 10000;
data.p->stop.store(true); data.p->stop.store(true);
data.p->nt | op::submit([p = data.p](auto) { data.p->tnt | op::submit([p = data.p](auto) {
::mi::set_done(p->out); ::mi::set_done(p->out);
}); });
}); });
nt | tnt |
op::submit([p, up = std::move(up)](auto nt) mutable { op::submit([p, up = std::move(up)](auto tnt) mutable {
// pass reference for cancellation. // pass reference for cancellation.
::mi::set_starting(p->out, std::move(up)); ::mi::set_starting(p->out, std::move(up));
}); });
...@@ -276,7 +183,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -276,7 +183,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
mi::on_starting([&](auto up) { mi::on_starting([&](auto up) {
signals += 10; signals += 10;
mi::set_next(up, 1); mi::set_next(up, 1);
nt | tcncl |
op::submit_at( op::submit_at(
at - 100ms, [up = std::move(up)](auto) mutable { at - 100ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up); ::mi::set_done(up);
...@@ -304,7 +211,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -304,7 +211,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
mi::on_starting([&](auto up) { mi::on_starting([&](auto up) {
signals += 10; signals += 10;
mi::set_next(up, 1); mi::set_next(up, 1);
nt | tcncl |
op::submit_at( op::submit_at(
at + 100ms, [up = std::move(up)](auto) mutable { at + 100ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up); ::mi::set_done(up);
...@@ -329,7 +236,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -329,7 +236,7 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
for (;;) { for (;;) {
signals = 0; signals = 0;
// set completion time to be in 100ms // set completion time to be in 100ms
at = nt.now() + 100ms; at = mi::now(tnt) + 100ms;
{ {
f | f |
op::blocking_submit( op::blocking_submit(
...@@ -340,14 +247,14 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -340,14 +247,14 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
mi::on_starting([&](auto up) { mi::on_starting([&](auto up) {
signals += 10; signals += 10;
mi::set_next(up, 1); mi::set_next(up, 1);
nt | op::submit_at(at, [up = std::move(up)](auto) mutable { tcncl | op::submit_at(at, [up = std::move(up)](auto) mutable {
::mi::set_done(up); ::mi::set_done(up);
}); });
})); }));
} }
// make sure any cancellation signal has completed // make sure any cancellation signal has completed
std::this_thread::sleep_for(10ms); std::this_thread::sleep_for(200ms);
// accumulate known signals // accumulate known signals
++total; ++total;
...@@ -373,6 +280,8 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") { ...@@ -373,6 +280,8 @@ SCENARIO("flow many cancellation new thread", "[flow][sender]") {
continue; continue;
} }
} }
time.join();
} }
} }
......
This diff is collapsed.
...@@ -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/trampoline.h"
#include "pushmi/new_thread.h" #include "pushmi/new_thread.h"
#include "pushmi/time_source.h"
using namespace pushmi::aliases; using namespace pushmi::aliases;
...@@ -40,21 +40,16 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) { ...@@ -40,21 +40,16 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) {
auto nt = v::new_thread(); auto nt = v::new_thread();
using NT = decltype(nt); using NT = decltype(nt);
// REQUIRE( v::TimeSingleDeferred< auto time = mi::time_source<>{};
// NT, v::archetype_single,
// NT&, std::exception_ptr> );
// REQUIRE( v::TimeExecutor<
// NT&, v::archetype_single,
// std::exception_ptr> );
auto any = v::make_any_time_executor(nt); auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })();
WHEN( "blocking submit now" ) { WHEN( "blocking submit now" ) {
auto signals = 0; auto signals = 0;
auto start = v::now(nt); auto start = v::now(tnt);
auto signaled = v::now(nt); auto signaled = start;
nt | tnt |
op::transform([](auto nt){ return nt | ep::now(); }) | op::transform([](auto tnt){ return tnt | ep::now(); }) |
op::blocking_submit( op::blocking_submit(
[&](auto at){ [&](auto at){
signaled = at; signaled = at;
...@@ -64,38 +59,44 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) { ...@@ -64,38 +59,44 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) {
THEN( "the value signal is recorded once and the signal did not drift much" ) { THEN( "the value signal is recorded once and the signal did not drift much" ) {
REQUIRE( signals == 100 ); REQUIRE( signals == 100 );
INFO("The delay is " << ::Catch::Detail::stringify(signaled - start)); auto delay = std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start)).count();
REQUIRE( signaled - start < 10s ); INFO("The delay is " << ::Catch::Detail::stringify(delay));
REQUIRE( delay < 1000 );
} }
} }
WHEN( "blocking get now" ) { WHEN( "blocking get now" ) {
auto start = v::now(nt); auto start = v::now(tnt);
auto signaled = nt | auto signaled = tnt |
op::transform([](auto nt){ op::transform([](auto tnt){
return v::now(nt); return v::now(tnt);
}) | }) |
op::get<std::chrono::system_clock::time_point>; op::get<std::chrono::system_clock::time_point>;
THEN( "the signal did not drift much" ) { THEN( "the signal did not drift much" ) {
INFO("The delay is " << ::Catch::Detail::stringify(signaled - start)); auto delay = std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start)).count();
REQUIRE( signaled - start < 10s ); INFO("The delay is " << ::Catch::Detail::stringify(delay));
REQUIRE( delay < 1000 );
} }
} }
WHEN( "submissions are ordered in time" ) { WHEN( "submissions are ordered in time" ) {
std::vector<std::string> times; std::vector<std::string> times;
std::atomic<int> pushed(0);
auto push = [&](int time) { auto push = [&](int time) {
return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); }); return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); ++pushed; });
}; };
nt | op::blocking_submit(v::on_value([push](auto nt) { tnt | op::submit(v::on_value([push](auto tnt) {
nt | auto now = tnt | ep::now();
tnt |
op::submit_after(40ms, push(40)) | op::submit_after(40ms, push(40)) |
op::submit_after(10ms, push(10)) | op::submit_at(now + 10ms, push(10)) |
op::submit_after(20ms, push(20)) | op::submit_after(20ms, push(20)) |
op::submit_after(10ms, push(11)); op::submit_at(now + 10ms, push(11));
})); }));
while(pushed.load() < 4) { std::this_thread::sleep_for(10ms); }
THEN( "the items were pushed in time order not insertion order" ) { THEN( "the items were pushed in time order not insertion order" ) {
REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"}); REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"});
} }
...@@ -103,9 +104,9 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) { ...@@ -103,9 +104,9 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) {
WHEN( "now is called" ) { WHEN( "now is called" ) {
bool done = false; bool done = false;
nt | ep::now(); tnt | ep::now();
nt | op::blocking_submit([&](auto nt) { tnt | op::blocking_submit([&](auto nt) {
nt | ep::now(); tnt | ep::now();
done = true; done = true;
}); });
...@@ -114,10 +115,37 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) { ...@@ -114,10 +115,37 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) {
} }
} }
WHEN( "blocking submit" ) {
auto signals = 0;
nt |
op::transform([](auto){ return 42; }) |
op::blocking_submit(
[&](auto){
signals += 100; },
[&](auto e) noexcept { signals += 1000; },
[&](){ signals += 10; });
THEN( "the value signal is recorded once" ) {
REQUIRE( signals == 100 );
}
}
WHEN( "blocking get" ) {
auto v = nt |
op::transform([](auto){
return 42;
}) |
op::get<int>;
THEN( "the result is" ) {
REQUIRE( v == 42 );
}
}
WHEN( "virtual derecursion is triggered" ) { WHEN( "virtual derecursion is triggered" ) {
int counter = 100'000; int counter = 100'000;
std::function<void(pushmi::any_time_executor_ref<> exec)> recurse; std::function<void(pushmi::any_executor_ref<> exec)> recurse;
recurse = [&](pushmi::any_time_executor_ref<> nt) { recurse = [&](pushmi::any_executor_ref<> nt) {
if (--counter <= 0) if (--counter <= 0)
return; return;
nt | op::submit(recurse); nt | op::submit(recurse);
...@@ -169,5 +197,7 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) { ...@@ -169,5 +197,7 @@ SCENARIO( "new_thread executor", "[new_thread][sender]" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"}); REQUIRE(values == std::vector<std::string>{"2.000000"});
} }
} }
time.join();
} }
} }
...@@ -36,83 +36,39 @@ struct countdownsingle { ...@@ -36,83 +36,39 @@ struct countdownsingle {
SCENARIO( "trampoline executor", "[trampoline][sender]" ) { SCENARIO( "trampoline executor", "[trampoline][sender]" ) {
GIVEN( "A trampoline time_single_sender" ) { GIVEN( "A trampoline single_sender" ) {
auto tr = v::trampoline(); auto tr = v::trampoline();
using TR = decltype(tr); using TR = decltype(tr);
// REQUIRE( v::TimeSingleDeferred< WHEN( "submit" ) {
// TR, v::archetype_single,
// TR&, std::exception_ptr> );
// REQUIRE( v::TimeExecutor<
// TR&, v::archetype_single,
// std::exception_ptr> );
WHEN( "submit now" ) {
auto signals = 0; auto signals = 0;
auto start = v::now(tr);
auto signaled = v::now(tr);
tr | tr |
op::transform([](auto tr){ return v::now(tr); }) | op::transform([](auto){ return 42; }) |
op::submit( op::submit(
[&](auto at){ signaled = at; [&](auto){
signals += 100; }, signals += 100; },
[&](auto e) noexcept { signals += 1000; }, [&](auto e) noexcept { signals += 1000; },
[&](){ signals += 10; }); [&](){ signals += 10; });
THEN( "the value signal is recorded once and the signal did not drift much" ) { THEN( "the value signal is recorded once" ) {
REQUIRE( signals == 100 ); REQUIRE( signals == 100 );
INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
REQUIRE( signaled - start < 10s );
}
}
WHEN( "blocking get now" ) {
auto start = v::now(tr);
auto signaled = tr |
op::transform([](auto tr){ return v::now(tr); }) |
op::get<decltype(v::now(tr))>;
THEN( "the signal did not drift much" ) {
INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
REQUIRE( signaled - start < 10s );
} }
} }
WHEN( "submissions are ordered in time" ) { WHEN( "blocking get" ) {
std::vector<std::string> times; auto v = tr |
auto push = [&](int time) { op::transform([](auto){ return 42; }) |
return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); }); op::get<int>;
};
tr | op::submit(v::on_value([push](auto tr) {
tr |
op::submit_after(40ms, push(40)) |
op::submit_after(10ms, push(10)) |
op::submit_after(20ms, push(20)) |
op::submit_after(10ms, push(11));
}));
THEN( "the items were pushed in time order not insertion order" ) {
REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"});
}
}
WHEN( "now is called" ) {
bool done = false;
tr | ep::now();
tr | op::submit([&](auto tr) {
tr | ep::now();
done = true;
});
THEN( "both calls to now() complete" ) { THEN( "the result is" ) {
REQUIRE( done == true ); REQUIRE( v == 42 );
} }
} }
WHEN( "virtual derecursion is triggered" ) { WHEN( "virtual derecursion is triggered" ) {
int counter = 100'000; int counter = 100'000;
std::function<void(pushmi::any_time_executor_ref<> exec)> recurse; std::function<void(pushmi::any_executor_ref<> exec)> recurse;
recurse = [&](pushmi::any_time_executor_ref<> tr) { recurse = [&](pushmi::any_executor_ref<> tr) {
if (--counter <= 0) if (--counter <= 0)
return; return;
tr | op::submit(recurse); tr | op::submit(recurse);
......
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