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

tests for flow single with cancellation

fbshipit-source-id: 57dbe1e4eac06af9fccc4e44ad269f27fccb70be
parent 7c22b071
...@@ -2161,6 +2161,9 @@ struct construct_deduced<none>; ...@@ -2161,6 +2161,9 @@ struct construct_deduced<none>;
template<> template<>
struct construct_deduced<single>; struct construct_deduced<single>;
template<>
struct construct_deduced<flow_single>;
template <template <class...> class T, class... AN> template <template <class...> class T, class... AN>
using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>; using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>;
...@@ -4529,6 +4532,14 @@ flow_single(Data d, DVF vf, DEF ef, DDF df, DStpF stpf, DStrtF strtf) ...@@ -4529,6 +4532,14 @@ flow_single(Data d, DVF vf, DEF ef, DDF df, DStpF stpf, DStrtF strtf)
template <class V, class PE = std::exception_ptr, class E = PE> template <class V, class PE = std::exception_ptr, class E = PE>
using any_flow_single = flow_single<V, PE, E>; using any_flow_single = flow_single<V, PE, E>;
template<>
struct construct_deduced<flow_single> {
template<class... AN>
auto operator()(AN&&... an) const -> decltype(pushmi::make_flow_single((AN&&) an...)) {
return pushmi::make_flow_single((AN&&) an...);
}
};
// template <class V, class PE = std::exception_ptr, class E = PE, class Wrapped> // template <class V, class PE = std::exception_ptr, class E = PE, class Wrapped>
// requires FlowSingleReceiver<Wrapped, V, PE, E> && !detail::is_v<Wrapped, none> && // requires FlowSingleReceiver<Wrapped, V, PE, E> && !detail::is_v<Wrapped, none> &&
// !detail::is_v<Wrapped, std::promise> // !detail::is_v<Wrapped, std::promise>
...@@ -4648,21 +4659,53 @@ class flow_single_deferred<SF> { ...@@ -4648,21 +4659,53 @@ class flow_single_deferred<SF> {
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_flow<>> && Invocable<SF&, Out>) (requires Receiver<Out, is_single<>, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
sf_(std::move(out)); sf_(std::move(out));
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data, class DSF>
class flow_single_deferred<Data, DSF> {
Data data_;
DSF sf_;
public:
using properties = property_set<is_sender<>, is_single<>>;
constexpr flow_single_deferred() = default;
constexpr explicit flow_single_deferred(Data data)
: data_(std::move(data)) {}
constexpr flow_single_deferred(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(defer::Receiver<Out, is_single<>, is_flow<>> PUSHMI_AND
defer::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
}
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_flow_single_deferred // make_flow_single_deferred
PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn { PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn {
inline auto operator()() const { inline auto operator()() const {
return flow_single_deferred<ignoreSF>{}; return flow_single_deferred<ignoreSF>{};
} }
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 flow_single_deferred<SF>(std::move(sf)); return flow_single_deferred<SF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d) const {
return flow_single_deferred<Data, passDSF>{std::move(d)};
}
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d, DSF sf) const {
return flow_single_deferred<Data, DSF>{std::move(d), std::move(sf)};
} }
} const make_flow_single_deferred {}; } const make_flow_single_deferred {};
...@@ -4671,8 +4714,17 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn { ...@@ -4671,8 +4714,17 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn {
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
flow_single_deferred() -> flow_single_deferred<ignoreSF>; flow_single_deferred() -> flow_single_deferred<ignoreSF>;
template <class SF> PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_single_deferred(SF) -> flow_single_deferred<SF>; flow_single_deferred(SF) -> flow_single_deferred<SF>;
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>)
flow_single_deferred(Data) -> flow_single_deferred<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>)
flow_single_deferred(Data, DSF) -> flow_single_deferred<Data, DSF>;
#endif #endif
template <class V, class PE = std::exception_ptr, class E = PE> template <class V, class PE = std::exception_ptr, class E = PE>
...@@ -5007,6 +5059,8 @@ inline auto new_thread() { ...@@ -5007,6 +5059,8 @@ inline auto new_thread() {
//#include "../deferred.h" //#include "../deferred.h"
//#include "../single_deferred.h" //#include "../single_deferred.h"
//#include "../time_single_deferred.h" //#include "../time_single_deferred.h"
//#include "../flow_single.h"
//#include "../flow_single_deferred.h"
//#include "../detail/if_constexpr.h" //#include "../detail/if_constexpr.h"
//#include "../detail/functional.h" //#include "../detail/functional.h"
...@@ -5038,17 +5092,20 @@ constexpr decltype(auto) apply(F&& f, Tuple&& t) { ...@@ -5038,17 +5092,20 @@ constexpr decltype(auto) apply(F&& f, Tuple&& t) {
namespace detail { namespace detail {
template <class Tag> template <class... TagN>
struct make_receiver; struct make_receiver;
template <> template <>
struct make_receiver<is_none<>> : construct_deduced<none> {}; struct make_receiver<is_none<>, void> : construct_deduced<none> {};
template <>
struct make_receiver<is_single<>, void> : construct_deduced<single> {};
template <> template <>
struct make_receiver<is_single<>> : construct_deduced<single> {}; struct make_receiver<is_single<>, is_flow<>> : construct_deduced<flow_single> {};
template <PUSHMI_TYPE_CONSTRAINT(Sender) In> template <PUSHMI_TYPE_CONSTRAINT(Sender) In>
struct out_from_fn { struct out_from_fn {
using Cardinality = property_set_index_t<properties_t<In>, is_silent<>>; using Cardinality = property_set_index_t<properties_t<In>, is_silent<>>;
using Make = make_receiver<Cardinality>; using Flow = std::conditional_t<property_query_v<properties_t<In>, is_flow<>>, is_flow<>, void>;
using Make = make_receiver<Cardinality, Flow>;
PUSHMI_TEMPLATE (class... Ts) PUSHMI_TEMPLATE (class... Ts)
(requires Invocable<Make, Ts...>) (requires Invocable<Make, Ts...>)
auto operator()(std::tuple<Ts...> args) const { auto operator()(std::tuple<Ts...> args) const {
...@@ -5288,7 +5345,12 @@ namespace submit_detail { ...@@ -5288,7 +5345,12 @@ namespace submit_detail {
template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN> template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN>
using receiver_type_t = using receiver_type_t =
pushmi::invoke_result_t< pushmi::invoke_result_t<
pushmi::detail::make_receiver<property_set_index_t<properties_t<In>, is_silent<>>>, pushmi::detail::make_receiver<
property_set_index_t<properties_t<In>, is_silent<>>,
std::conditional_t<
property_query_v<properties_t<In>, is_flow<>>,
is_flow<>,
void>>,
AN...>; AN...>;
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
......
...@@ -34,6 +34,9 @@ struct construct_deduced<none>; ...@@ -34,6 +34,9 @@ struct construct_deduced<none>;
template<> template<>
struct construct_deduced<single>; struct construct_deduced<single>;
template<>
struct construct_deduced<flow_single>;
template <template <class...> class T, class... AN> template <template <class...> class T, class... AN>
using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>; using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>;
......
...@@ -531,6 +531,14 @@ flow_single(Data d, DVF vf, DEF ef, DDF df, DStpF stpf, DStrtF strtf) ...@@ -531,6 +531,14 @@ flow_single(Data d, DVF vf, DEF ef, DDF df, DStpF stpf, DStrtF strtf)
template <class V, class PE = std::exception_ptr, class E = PE> template <class V, class PE = std::exception_ptr, class E = PE>
using any_flow_single = flow_single<V, PE, E>; using any_flow_single = flow_single<V, PE, E>;
template<>
struct construct_deduced<flow_single> {
template<class... AN>
auto operator()(AN&&... an) const -> decltype(pushmi::make_flow_single((AN&&) an...)) {
return pushmi::make_flow_single((AN&&) an...);
}
};
// template <class V, class PE = std::exception_ptr, class E = PE, class Wrapped> // template <class V, class PE = std::exception_ptr, class E = PE, class Wrapped>
// requires FlowSingleReceiver<Wrapped, V, PE, E> && !detail::is_v<Wrapped, none> && // requires FlowSingleReceiver<Wrapped, V, PE, E> && !detail::is_v<Wrapped, none> &&
// !detail::is_v<Wrapped, std::promise> // !detail::is_v<Wrapped, std::promise>
......
...@@ -109,21 +109,53 @@ class flow_single_deferred<SF> { ...@@ -109,21 +109,53 @@ class flow_single_deferred<SF> {
: sf_(std::move(sf)) {} : sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_flow<>> && Invocable<SF&, Out>) (requires Receiver<Out, is_single<>, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) { void submit(Out out) {
sf_(std::move(out)); sf_(std::move(out));
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data, class DSF>
class flow_single_deferred<Data, DSF> {
Data data_;
DSF sf_;
public:
using properties = property_set<is_sender<>, is_single<>>;
constexpr flow_single_deferred() = default;
constexpr explicit flow_single_deferred(Data data)
: data_(std::move(data)) {}
constexpr flow_single_deferred(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(defer::Receiver<Out, is_single<>, is_flow<>> PUSHMI_AND
defer::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
}
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// make_flow_single_deferred // make_flow_single_deferred
PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn { PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn {
inline auto operator()() const { inline auto operator()() const {
return flow_single_deferred<ignoreSF>{}; return flow_single_deferred<ignoreSF>{};
} }
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 flow_single_deferred<SF>(std::move(sf)); return flow_single_deferred<SF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d) const {
return flow_single_deferred<Data, passDSF>{std::move(d)};
}
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>)
auto operator()(Data d, DSF sf) const {
return flow_single_deferred<Data, DSF>{std::move(d), std::move(sf)};
} }
} const make_flow_single_deferred {}; } const make_flow_single_deferred {};
...@@ -132,8 +164,17 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn { ...@@ -132,8 +164,17 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_deferred_fn {
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
flow_single_deferred() -> flow_single_deferred<ignoreSF>; flow_single_deferred() -> flow_single_deferred<ignoreSF>;
template <class SF> PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_single_deferred(SF) -> flow_single_deferred<SF>; flow_single_deferred(SF) -> flow_single_deferred<SF>;
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_single<>, is_flow<>>)
flow_single_deferred(Data) -> flow_single_deferred<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_single<>, is_flow<>>)
flow_single_deferred(Data, DSF) -> flow_single_deferred<Data, DSF>;
#endif #endif
template <class V, class PE = std::exception_ptr, class E = PE> template <class V, class PE = std::exception_ptr, class E = PE>
......
...@@ -13,6 +13,8 @@ ...@@ -13,6 +13,8 @@
#include "../deferred.h" #include "../deferred.h"
#include "../single_deferred.h" #include "../single_deferred.h"
#include "../time_single_deferred.h" #include "../time_single_deferred.h"
#include "../flow_single.h"
#include "../flow_single_deferred.h"
#include "../detail/if_constexpr.h" #include "../detail/if_constexpr.h"
#include "../detail/functional.h" #include "../detail/functional.h"
...@@ -44,17 +46,20 @@ constexpr decltype(auto) apply(F&& f, Tuple&& t) { ...@@ -44,17 +46,20 @@ constexpr decltype(auto) apply(F&& f, Tuple&& t) {
namespace detail { namespace detail {
template <class Tag> template <class... TagN>
struct make_receiver; struct make_receiver;
template <> template <>
struct make_receiver<is_none<>> : construct_deduced<none> {}; struct make_receiver<is_none<>, void> : construct_deduced<none> {};
template <> template <>
struct make_receiver<is_single<>> : construct_deduced<single> {}; struct make_receiver<is_single<>, void> : construct_deduced<single> {};
template <>
struct make_receiver<is_single<>, is_flow<>> : construct_deduced<flow_single> {};
template <PUSHMI_TYPE_CONSTRAINT(Sender) In> template <PUSHMI_TYPE_CONSTRAINT(Sender) In>
struct out_from_fn { struct out_from_fn {
using Cardinality = property_set_index_t<properties_t<In>, is_silent<>>; using Cardinality = property_set_index_t<properties_t<In>, is_silent<>>;
using Make = make_receiver<Cardinality>; using Flow = std::conditional_t<property_query_v<properties_t<In>, is_flow<>>, is_flow<>, void>;
using Make = make_receiver<Cardinality, Flow>;
PUSHMI_TEMPLATE (class... Ts) PUSHMI_TEMPLATE (class... Ts)
(requires Invocable<Make, Ts...>) (requires Invocable<Make, Ts...>)
auto operator()(std::tuple<Ts...> args) const { auto operator()(std::tuple<Ts...> args) const {
......
...@@ -20,7 +20,12 @@ namespace submit_detail { ...@@ -20,7 +20,12 @@ namespace submit_detail {
template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN> template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN>
using receiver_type_t = using receiver_type_t =
pushmi::invoke_result_t< pushmi::invoke_result_t<
pushmi::detail::make_receiver<property_set_index_t<properties_t<In>, is_silent<>>>, pushmi::detail::make_receiver<
property_set_index_t<properties_t<In>, is_silent<>>,
std::conditional_t<
property_query_v<properties_t<In>, is_flow<>>,
is_flow<>,
void>>,
AN...>; AN...>;
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
......
...@@ -2,6 +2,7 @@ ...@@ -2,6 +2,7 @@
add_executable(PushmiTest add_executable(PushmiTest
catch.cpp catch.cpp
FlowTest.cpp
CompileTest.cpp CompileTest.cpp
NewThreadTest.cpp NewThreadTest.cpp
TrampolineTest.cpp TrampolineTest.cpp
...@@ -13,4 +14,4 @@ target_link_libraries(PushmiTest ...@@ -13,4 +14,4 @@ target_link_libraries(PushmiTest
) )
include(../external/Catch2/contrib/Catch.cmake) include(../external/Catch2/contrib/Catch.cmake)
catch_discover_tests(PushmiTest) catch_discover_tests(PushmiTest)
\ No newline at end of file
#include "catch.hpp"
#include <type_traits>
#include <chrono>
using namespace std::literals;
#include "pushmi/flow_single_deferred.h"
#include "pushmi/o/submit.h"
#include "pushmi/trampoline.h"
#include "pushmi/new_thread.h"
using namespace pushmi::aliases;
#if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_
#define MAKE_(...) {__VA_ARGS__}
#else
#define MAKE(x) make_ ## x
#endif
SCENARIO( "flow single immediate cancellation", "[flow][deferred]" ) {
int signals = 0;
GIVEN( "A flow single deferred" ) {
auto f = mi::MAKE(flow_single_deferred)([&](auto out){
// boolean cancellation - on stack
bool stop = false;
auto up = mi::MAKE(none)(
[&](auto e) noexcept {signals += 1000000; stop = true;},
[&](){signals += 100000; stop = true;});
// pass reference for cancellation.
::mi::set_starting(out, up);
// check boolean to select signal
if (!stop) {
::mi::set_value(out, 42);
} else {
// cancellation is not an error
::mi::set_done(out);
}
// I want to get rid of this signal it makes usage harder and
// messes up r-value qualifing done, error and value.
::mi::set_stopping(out);
});
WHEN( "submit is applied and cancels the producer" ) {
f | op::submit(
mi::on_value([&](int){ signals = 100; }),
mi::on_error([&](auto) noexcept { signals = 1000; }),
mi::on_done([&](){signals += 1;}),
mi::on_stopping([&](){signals += 10000;}),
// immediately stop producer
mi::on_starting([&](auto up){ signals = 10; ::mi::set_done(up); }));
THEN( "the starting, up.done, out.done out.stopping signals are each recorded once" ) {
REQUIRE(signals == 110011);
}
}
}
}
SCENARIO( "flow single cancellation", "[flow][deferred]" ) {
auto tr = v::trampoline();
using TR = decltype(tr);
int signals = 0;
GIVEN( "A flow single deferred" ) {
auto f = mi::MAKE(flow_single_deferred)([&](auto out){
// boolean cancellation - on stack
bool stop = false;
auto up = mi::MAKE(none)(
[&](auto e) noexcept {signals += 1000000; stop = true;},
[&](){signals += 100000; stop = true;});
// blocking the stack to keep 'up' alive
tr | op::blocking_submit([&](auto tr){
// pass reference for cancellation.
::mi::set_starting(out, up);
tr | op::submit_after(
100ms,
[&](auto) {
// check boolean to select signal
if (!stop) {
::mi::set_value(out, 42);
} else {
// cancellation is not an error
::mi::set_done(out);
}
// I want to get rid of this signal it makes usage harder and
// messes up r-value qualifing done, error and value.
::mi::set_stopping(out);
}
);
});
});
WHEN( "submit is applied and cancels the producer" ) {
f | op::submit(
mi::on_value([&](int){ signals = 100; }),
mi::on_error([&](auto) noexcept { signals = 1000; }),
mi::on_done([&](){signals += 1;}),
mi::on_stopping([&](){signals += 10000;}),
// stop producer before it is scheduled to run
mi::on_starting([&](auto up){ signals = 10; tr | op::submit_after(50ms, [up](auto) mutable {::mi::set_done(up);}); }));
THEN( "the starting, up.done, out.done out.stopping signals are each recorded once" ) {
REQUIRE(signals == 110011);
}
}
}
}
...@@ -60,7 +60,7 @@ SCENARIO( "empty can be used with tap and submit", "[empty][deferred]" ) { ...@@ -60,7 +60,7 @@ SCENARIO( "empty can be used with tap and submit", "[empty][deferred]" ) {
REQUIRE( v::SenderTo<E, v::any_single<int>, v::is_single<>> ); REQUIRE( v::SenderTo<E, v::any_single<int>, v::is_single<>> );
WHEN( "tap and submit are applied" ) { WHEN( "tap and submit are applied" ) {
int signals = 0; int signals = 0;
e | e |
op::tap( op::tap(
......
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