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

migrate to gtest

Summary:
migrated tests
fixed some bugs introduced while tests were disabled

Reviewed By: yfeldblum

Differential Revision: D10515231

fbshipit-source-id: 34cddb6588c4da9cd05256111bf9626b013dd4fa
parent 20d6c1df
...@@ -84,8 +84,8 @@ class any_constrained_single_sender { ...@@ -84,8 +84,8 @@ class any_constrained_single_sender {
return ::folly::pushmi::top(*static_cast<Wrapped*>((void*)src.buffer_)); return ::folly::pushmi::top(*static_cast<Wrapped*>((void*)src.buffer_));
} }
static any_constrained_executor<E, CV> executor(data& src) { static any_constrained_executor<E, CV> executor(data& src) {
return any_constrained_executor<E, CV>{ return any_constrained_executor<E, CV>{::folly::pushmi::executor(
executor(*static_cast<Wrapped*>((void*)src.buffer_))}; *static_cast<Wrapped*>((void*)src.buffer_))};
} }
static void submit(data& src, CV cv, any_receiver<E, VN...> out) { static void submit(data& src, CV cv, any_receiver<E, VN...> out) {
::folly::pushmi::submit( ::folly::pushmi::submit(
......
...@@ -166,7 +166,7 @@ struct entangled { ...@@ -166,7 +166,7 @@ struct entangled {
entangled& operator=(entangled&&) = delete; entangled& operator=(entangled&&) = delete;
explicit entangled(T t) explicit entangled(T t)
: t(std::move(t)), dual(nullptr), stateMachine(kUnlocked) {} : stateMachine(kUnlocked), t(std::move(t)), dual(nullptr) {}
entangled(entangled&& other) entangled(entangled&& other)
: stateMachine((other.lockBoth(), kLocked)), : stateMachine((other.lockBoth(), kLocked)),
t(std::move(other.t)), t(std::move(other.t)),
......
...@@ -34,16 +34,27 @@ struct for_each_fn { ...@@ -34,16 +34,27 @@ struct for_each_fn {
using properties = using properties =
property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>; property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>;
std::function<void(std::ptrdiff_t)> pull; std::function<void(std::ptrdiff_t)> pull;
template <class V> template <class... VN>
void value(V&& v) { void value(VN&&... vn) {
set_value(static_cast<Out&>(*this), (V &&) v); ::folly::pushmi::set_value(static_cast<Out&>(*this), (VN &&) vn...);
pull(1); pull(1);
} }
template <class E>
void error(E&& e) {
// break circular reference
pull = nullptr;
::folly::pushmi::set_error(static_cast<Out&>(*this), (E &&) e);
}
void done() {
// break circular reference
pull = nullptr;
::folly::pushmi::set_done(static_cast<Out&>(*this));
}
PUSHMI_TEMPLATE(class Up) PUSHMI_TEMPLATE(class Up)
(requires Receiver<Up>) (requires Receiver<Up> && ReceiveValue<Up, std::ptrdiff_t>)
void starting(Up up) { void starting(Up up) {
pull = [up = std::move(up)](std::ptrdiff_t requested) mutable { pull = [up = std::move(up)](std::ptrdiff_t requested) mutable {
set_value(up, requested); ::folly::pushmi::set_value(up, requested);
}; };
pull(1); pull(1);
} }
...@@ -62,23 +73,8 @@ struct for_each_fn { ...@@ -62,23 +73,8 @@ struct for_each_fn {
property_set_index_t<properties_t<In>, is_single<>>>>()( property_set_index_t<properties_t<In>, is_single<>>>>()(
std::move(args_))}; std::move(args_))};
using Out = decltype(out); using Out = decltype(out);
submit( ::folly::pushmi::submit(
in,
::folly::pushmi::detail::receiver_from_fn<In>()(
Pull<In, Out>{std::move(out)}));
return in;
}
PUSHMI_TEMPLATE(class In)
(requires Sender<In>&& Constrained<In>&& Flow<In>&& Many<In>)
In operator()(In in) {
auto out{::folly::pushmi::detail::receiver_from_fn<subset<
is_sender<>,
property_set_index_t<properties_t<In>, is_single<>>>>()(
std::move(args_))};
using Out = decltype(out);
submit(
in, in,
::folly::pushmi::top(in),
::folly::pushmi::detail::receiver_from_fn<In>()( ::folly::pushmi::detail::receiver_from_fn<In>()(
Pull<In, Out>{std::move(out)})); Pull<In, Out>{std::move(out)}));
return in; return in;
...@@ -88,7 +84,7 @@ struct for_each_fn { ...@@ -88,7 +84,7 @@ struct for_each_fn {
public: public:
template <class... AN> template <class... AN>
auto operator()(AN&&... an) const { auto operator()(AN&&... an) const {
return for_each_fn::fn<AN...>{{(AN &&) an...}}; return for_each_fn::fn<AN...>{std::tuple<AN...>{(AN &&) an...}};
} }
}; };
......
...@@ -102,32 +102,35 @@ struct flow_from_up { ...@@ -102,32 +102,35 @@ struct flow_from_up {
return; return;
} }
// submit work to exec // submit work to exec
submit(p->exec, make_receiver([p = p, requested](auto) { ::folly::pushmi::submit(
auto remaining = requested; p->exec, make_receiver([p = p, requested](auto) {
// this loop is structured to work when there is auto remaining = requested;
// re-entrancy out.value in the loop may call up.value. // this loop is structured to work when there is
// to handle this the state of p->c must be captured and // re-entrancy out.value in the loop may call up.value.
// the remaining and p->c must be changed before // to handle this the state of p->c must be captured and
// out.value is called. // the remaining and p->c must be changed before
while (remaining-- > 0 && !p->stop && p->c != p->end) { // out.value is called.
auto i = (p->c)++; while (remaining-- > 0 && !p->stop && p->c != p->end) {
set_value(p->out, ::folly::pushmi::detail::as_const(*i)); auto i = (p->c)++;
} set_value(p->out, ::folly::pushmi::detail::as_const(*i));
if (p->c == p->end) { }
set_done(p->out); if (p->c == p->end) {
} set_done(p->out);
})); }
}));
} }
template <class E> template <class E>
void error(E) noexcept { void error(E) noexcept {
p->stop.store(true); p->stop.store(true);
submit(p->exec, make_receiver([p = p](auto) { set_done(p->out); })); ::folly::pushmi::submit(
p->exec, make_receiver([p = p](auto) { set_done(p->out); }));
} }
void done() { void done() {
p->stop.store(true); p->stop.store(true);
submit(p->exec, make_receiver([p = p](auto) { set_done(p->out); })); ::folly::pushmi::submit(
p->exec, make_receiver([p = p](auto) { set_done(p->out); }));
} }
}; };
...@@ -147,10 +150,11 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn { ...@@ -147,10 +150,11 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn {
auto p = std::make_shared<Producer>( auto p = std::make_shared<Producer>(
begin_, end_, std::move(out), exec_, false); begin_, end_, std::move(out), exec_, false);
submit(exec_, make_receiver([p](auto) { ::folly::pushmi::submit(
// pass reference for cancellation. exec_, make_receiver([p](auto) {
set_starting(p->out, make_receiver(flow_from_up<Producer>{p})); // pass reference for cancellation.
})); set_starting(p->out, make_receiver(flow_from_up<Producer>{p}));
}));
} }
}; };
......
...@@ -58,7 +58,7 @@ struct submit_fn { ...@@ -58,7 +58,7 @@ struct submit_fn {
In operator()(In in) { In operator()(In in) {
auto out{ auto out{
::folly::pushmi::detail::receiver_from_fn<In>{}(std::move(args_))}; ::folly::pushmi::detail::receiver_from_fn<In>{}(std::move(args_))};
submit(in, std::move(out)); ::folly::pushmi::submit(in, std::move(out));
return in; return in;
} }
}; };
...@@ -81,7 +81,7 @@ struct submit_at_fn { ...@@ -81,7 +81,7 @@ struct submit_at_fn {
In operator()(In in) { In operator()(In in) {
auto out{ auto out{
::folly::pushmi::detail::receiver_from_fn<In>()(std::move(args_))}; ::folly::pushmi::detail::receiver_from_fn<In>()(std::move(args_))};
submit(in, std::move(at_), std::move(out)); ::folly::pushmi::submit(in, std::move(at_), std::move(out));
return in; return in;
} }
}; };
...@@ -90,7 +90,8 @@ struct submit_at_fn { ...@@ -90,7 +90,8 @@ struct submit_at_fn {
PUSHMI_TEMPLATE(class TP, class... AN) PUSHMI_TEMPLATE(class TP, class... AN)
(requires Regular<TP>) (requires Regular<TP>)
auto operator()(TP at, AN... an) const { auto operator()(TP at, AN... an) const {
return submit_at_fn::fn<TP, AN...>{std::move(at), {(AN &&) an...}}; return submit_at_fn::fn<TP, AN...>{std::move(at),
std::tuple<AN...>{(AN &&) an...}};
} }
}; };
...@@ -109,7 +110,7 @@ struct submit_after_fn { ...@@ -109,7 +110,7 @@ struct submit_after_fn {
auto out{ auto out{
::folly::pushmi::detail::receiver_from_fn<In>()(std::move(args_))}; ::folly::pushmi::detail::receiver_from_fn<In>()(std::move(args_))};
auto at = ::folly::pushmi::now(in) + std::move(after_); auto at = ::folly::pushmi::now(in) + std::move(after_);
submit(in, std::move(at), std::move(out)); ::folly::pushmi::submit(in, std::move(at), std::move(out));
return in; return in;
} }
}; };
...@@ -118,7 +119,8 @@ struct submit_after_fn { ...@@ -118,7 +119,8 @@ struct submit_after_fn {
PUSHMI_TEMPLATE(class D, class... AN) PUSHMI_TEMPLATE(class D, class... AN)
(requires Regular<D>) (requires Regular<D>)
auto operator()(D after, AN... an) const { auto operator()(D after, AN... an) const {
return submit_after_fn::fn<D, AN...>{std::move(after), {(AN &&) an...}}; return submit_after_fn::fn<D, AN...>{std::move(after),
std::tuple<AN...>{(AN &&) an...}};
} }
}; };
...@@ -159,7 +161,7 @@ struct blocking_submit_fn { ...@@ -159,7 +161,7 @@ struct blocking_submit_fn {
using properties = properties_t<Exec>; using properties = properties_t<Exec>;
auto executor() { auto executor() {
return make(state_, executor(ex_)); return make(state_, ::folly::pushmi::executor(ex_));
} }
PUSHMI_TEMPLATE(class... ZN) PUSHMI_TEMPLATE(class... ZN)
...@@ -172,14 +174,16 @@ struct blocking_submit_fn { ...@@ -172,14 +174,16 @@ struct blocking_submit_fn {
(requires Receiver<Out>&& Constrained<Exec>) (requires Receiver<Out>&& Constrained<Exec>)
void submit(CV cv, Out out) { void submit(CV cv, Out out) {
++state_->nested; ++state_->nested;
submit(ex_, cv, nested_receiver_impl<Out>{state_, std::move(out)}); ::folly::pushmi::submit(
ex_, cv, nested_receiver_impl<Out>{state_, std::move(out)});
} }
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out> && not Constrained<Exec>) (requires Receiver<Out> && not Constrained<Exec>)
void submit(Out out) { void submit(Out out) {
++state_->nested; ++state_->nested;
submit(ex_, nested_receiver_impl<Out>{state_, std::move(out)}); ::folly::pushmi::submit(
ex_, nested_receiver_impl<Out>{state_, std::move(out)});
} }
}; };
template <class Out> template <class Out>
...@@ -293,7 +297,7 @@ struct blocking_submit_fn { ...@@ -293,7 +297,7 @@ struct blocking_submit_fn {
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>&& SenderTo<In, Out>) (requires Receiver<Out>&& SenderTo<In, Out>)
void operator()(In& in, Out out) const { void operator()(In& in, Out out) const {
submit(in, std::move(out)); ::folly::pushmi::submit(in, std::move(out));
} }
}; };
// TODO - only move, move-only types.. // TODO - only move, move-only types..
......
...@@ -203,7 +203,8 @@ class strand_executor { ...@@ -203,7 +203,8 @@ class strand_executor {
queue_->items_.push(any_receiver<E, any_executor_ref<E>>{std::move(out)}); queue_->items_.push(any_receiver<E, any_executor_ref<E>>{std::move(out)});
if (queue_->remaining_ == 0) { if (queue_->remaining_ == 0) {
// noone is minding the shop, send a worker // noone is minding the shop, send a worker
submit(queue_->ex_, strand_queue_receiver<E, Executor>{queue_}); ::folly::pushmi::submit(
queue_->ex_, strand_queue_receiver<E, Executor>{queue_});
} }
} }
}; };
......
...@@ -25,6 +25,11 @@ ...@@ -25,6 +25,11 @@
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using namespace testing;
using namespace std::literals; using namespace std::literals;
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
...@@ -562,3 +567,5 @@ void flow_many_sender_test() { ...@@ -562,3 +567,5 @@ void flow_many_sender_test() {
mi::Executor<mi::executor_t<decltype(in0)>>, mi::Executor<mi::executor_t<decltype(in0)>>,
"sender has invalid executor"); "sender has invalid executor");
} }
TEST(CompileTest, Test) {}
...@@ -22,9 +22,9 @@ ...@@ -22,9 +22,9 @@
using namespace std::literals; using namespace std::literals;
#include <folly/experimental/pushmi/flow_many_sender.h> #include <folly/experimental/pushmi/flow_many_sender.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/from.h>
#include <folly/experimental/pushmi/o/for_each.h> #include <folly/experimental/pushmi/o/for_each.h>
#include <folly/experimental/pushmi/o/from.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/entangle.h> #include <folly/experimental/pushmi/entangle.h>
#include <folly/experimental/pushmi/new_thread.h> #include <folly/experimental/pushmi/new_thread.h>
...@@ -33,7 +33,10 @@ using namespace std::literals; ...@@ -33,7 +33,10 @@ using namespace std::literals;
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#if 0 #include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using namespace testing;
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_ #define MAKE(x) x MAKE_
...@@ -43,12 +46,10 @@ using namespace folly::pushmi::aliases; ...@@ -43,12 +46,10 @@ using namespace folly::pushmi::aliases;
#define MAKE(x) make_##x #define MAKE(x) make_##x
#endif #endif
SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") { class ImmediateFlowManySender : public Test {
int signals = 0; protected:
auto make_producer() {
GIVEN("A flow many sender") { return mi::MAKE(flow_many_sender)([&](auto out) {
auto f = mi::MAKE(flow_many_sender)([&](auto out) {
using Out = decltype(out); using Out = decltype(out);
struct Data : mi::receiver<> { struct Data : mi::receiver<> {
explicit Data(Out out) : out(std::move(out)), stop(false) {} explicit Data(Out out) : out(std::move(out)), stop(false) {}
...@@ -59,21 +60,23 @@ SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") { ...@@ -59,21 +60,23 @@ SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") {
auto up = mi::MAKE(receiver)( auto up = mi::MAKE(receiver)(
Data{std::move(out)}, Data{std::move(out)},
[&](auto& data, auto requested) { [&](auto& data, auto requested) {
signals += 1000000; signals_ += 1000000;
if (requested < 1) {return;} if (requested < 1) {
return;
}
// check boolean to select signal // check boolean to select signal
if (!data.stop) { if (!data.stop) {
::mi::set_value(data.out, 42); ::mi::set_value(data.out, 42);
} }
::mi::set_done(data.out); ::mi::set_done(data.out);
}, },
[&](auto& data, auto e) noexcept { [&](auto& data, auto) noexcept {
signals += 100000; signals_ += 100000;
data.stop = true; data.stop = true;
::mi::set_done(data.out); ::mi::set_done(data.out);
}, },
[&](auto& data) { [&](auto& data) {
signals += 10000; signals_ += 10000;
data.stop = true; data.stop = true;
::mi::set_done(data.out); ::mi::set_done(data.out);
}); });
...@@ -81,68 +84,78 @@ SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") { ...@@ -81,68 +84,78 @@ SCENARIO("flow many immediate cancellation", "[flowmany][flow][sender]") {
// pass reference for cancellation. // pass reference for cancellation.
::mi::set_starting(up.data().out, std::move(up)); ::mi::set_starting(up.data().out, std::move(up));
}); });
}
WHEN("submit is applied and cancels the producer") { template <class F>
f | auto make_consumer(F f) {
op::submit(mi::MAKE(flow_receiver)( return mi::MAKE(flow_receiver)(
mi::on_value([&](int) { signals += 100; }), mi::on_value([&](int) { signals_ += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }), mi::on_error([&](auto) noexcept { signals_ += 1000; }),
mi::on_done([&]() { signals += 1; }), mi::on_done([&]() { signals_ += 1; }),
// immediately stop producer mi::on_starting([&, f](auto up) {
mi::on_starting([&](auto up) { signals_ += 10;
signals += 10; f(std::move(up));
::mi::set_done(up); }));
}))); }
THEN( int signals_{0};
"the starting, up.done and out.done signals are each recorded once") { };
REQUIRE(signals == 10011);
}
}
WHEN("submit is applied and cancels the producer late") { TEST_F(ImmediateFlowManySender, EarlyCancellation) {
f | make_producer() | op::submit(make_consumer([](auto up) {
op::submit(mi::MAKE(flow_receiver)( // immediately stop producer
mi::on_value([&](int) { signals += 100; }), ::mi::set_done(up);
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_value(up, 1);
})));
THEN(
"the starting, up.value, value and done signals are each recorded once") {
REQUIRE(signals == 1000111);
}
}
} EXPECT_THAT(signals_, Eq(10011))
<< "expected that the starting, up.done and out.done signals are each recorded once";
} }
SCENARIO("flow many cancellation new thread", "[flowmany][flow][sender]") { TEST_F(ImmediateFlowManySender, LateCancellation) {
auto nt = mi::new_thread(); make_producer() | op::submit(make_consumer([](auto up) {
using NT = decltype(nt); // do not stop producer before it is scheduled to run
auto time = mi::time_source<>{}; ::mi::set_value(up, 1);
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })(); }));
using TNT = decltype(tnt);
auto tcncl = time.make(mi::systemNowF{}, [nt](){ return nt; })(); EXPECT_THAT(signals_, Eq(1000111))
std::atomic<int> signals{0}; << "expected that the starting, up.value, value and done signals are each recorded once";
auto at = mi::now(tnt) + 200ms; }
GIVEN("A flow many sender") { using NT = decltype(mi::new_thread());
inline auto make_time(mi::time_source<>& t, NT& ex) {
return t.make(mi::systemNowF{}, [ex]() { return ex; })();
}
class ConcurrentFlowManySender : public Test {
protected:
using TNT = mi::invoke_result_t<decltype(make_time), mi::time_source<>&, NT&>;
void reset() {
at_ = mi::now(tnt_) + 100ms;
signals_ = 0;
terminal_ = 0;
cancel_ = 0;
}
void join() {
timeproduce_.join();
timecancel_.join();
}
void cancellation_test(std::chrono::system_clock::time_point at) {
auto f = mi::MAKE(flow_many_sender)([&](auto out) { auto f = mi::MAKE(flow_many_sender)([&](auto out) {
using Out = decltype(out); using Out = decltype(out);
// boolean cancellation // boolean cancellation
struct producer { struct producer {
producer(Out out, TNT tnt, bool s) : out(std::move(out)), tnt(std::move(tnt)), stop(s) {} producer(Out out, TNT tnt, bool s)
: out(std::move(out)), tnt(std::move(tnt)), stop(s) {}
Out out; Out out;
TNT tnt; TNT tnt;
std::atomic<bool> stop; std::atomic<bool> stop;
}; };
auto p = std::make_shared<producer>(std::move(out), tnt, false); auto p = std::make_shared<producer>(std::move(out), tnt_, false);
struct Data : mi::receiver<> { struct Data : mi::receiver<> {
explicit Data(std::shared_ptr<producer> p) : p(std::move(p)) {} explicit Data(std::shared_ptr<producer> p) : p(std::move(p)) {}
...@@ -151,169 +164,138 @@ SCENARIO("flow many cancellation new thread", "[flowmany][flow][sender]") { ...@@ -151,169 +164,138 @@ SCENARIO("flow many cancellation new thread", "[flowmany][flow][sender]") {
auto up = mi::MAKE(receiver)( auto up = mi::MAKE(receiver)(
Data{p}, Data{p},
[&at, &signals](auto& data, auto requested) { [&](auto& data, auto requested) {
signals += 1000000; signals_ += 1000000;
if (requested < 1) {return;} if (requested < 1) {
return;
}
// submit work to happen later // submit work to happen later
data.p->tnt | data.p->tnt | op::submit_at(at_, [p = data.p](auto) {
op::submit_at( // check boolean to select signal
at, if (!p->stop) {
[p = data.p](auto) { ::mi::set_value(p->out, 42);
// check boolean to select signal ::mi::set_done(p->out);
if (!p->stop) { }
::mi::set_value(p->out, 42); });
}
::mi::set_done(p->out);
});
}, },
[&signals](auto& data, auto e) noexcept { [&](auto& data, auto) noexcept {
signals += 100000; signals_ += 100000;
data.p->stop.store(true); data.p->stop.store(true);
data.p->tnt | op::submit([p = data.p](auto) { data.p->tnt |
::mi::set_done(p->out); op::submit([p = data.p](auto) { ::mi::set_done(p->out); });
}); ++cancel_;
}, },
[&signals](auto& data) { [&](auto& data) {
signals += 10000; signals_ += 10000;
data.p->stop.store(true); data.p->stop.store(true);
data.p->tnt | op::submit([p = data.p](auto) { data.p->tnt |
::mi::set_done(p->out); op::submit([p = data.p](auto) { ::mi::set_done(p->out); });
}); ++cancel_;
}); });
tnt | tnt_ | op::submit([p, sup = std::move(up)](auto) 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(sup));
::mi::set_starting(p->out, std::move(up)); });
});
}); });
f |
op::submit(mi::MAKE(flow_receiver)(
mi::on_value([&](int) { signals_ += 100; }),
mi::on_error([&](auto) noexcept {
signals_ += 1000;
++terminal_;
}),
mi::on_done([&]() {
signals_ += 1;
++terminal_;
}),
// stop producer before it is scheduled to run
mi::on_starting([&, at](auto up) {
signals_ += 10;
mi::set_value(up, 1);
tcncl_ | op::submit_at(at, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
})));
WHEN("submit is applied and cancels the producer early") { while (terminal_ == 0 || cancel_ == 0) {
{ std::this_thread::yield();
f |
op::blocking_submit(
mi::on_value([&](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;
mi::set_value(up, 1);
tcncl |
op::submit_at(
at - 100ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
// make sure that the completion signal arrives
std::this_thread::sleep_for(200ms);
THEN(
"the starting, up.done and out.done signals are each recorded once") {
REQUIRE(signals == 1010011);
}
} }
}
WHEN("submit is applied and cancels the producer late") { NT ntproduce_{mi::new_thread()};
{ mi::time_source<> timeproduce_{};
f | TNT tnt_{make_time(timeproduce_, ntproduce_)};
op::blocking_submit(
mi::on_value([&](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_value(up, 1);
tcncl |
op::submit_at(
at + 100ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
std::this_thread::sleep_for(200ms);
THEN(
"the starting, up.done and out.value signals are each recorded once") {
REQUIRE(signals == 1010111);
}
}
WHEN("submit is applied and cancels the producer at the same time") { NT ntcancel_{mi::new_thread()};
// count known results mi::time_source<> timecancel_{};
int total = 0; TNT tcncl_{make_time(timecancel_, ntcancel_)};
int cancellostrace = 0; // 1010111
int cancelled = 0; // 1010011
for (;;) {
signals = 0;
// set completion time to be in 100ms
at = mi::now(tnt) + 100ms;
{
f |
op::blocking_submit(
mi::on_value([&](int) { signals += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }),
mi::on_done([&]() { signals += 1; }),
// stop producer at the same time that it is scheduled to run
mi::on_starting([&](auto up) {
signals += 10;
mi::set_value(up, 1);
tcncl | op::submit_at(at, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
// make sure any cancellation signal has completed
std::this_thread::sleep_for(200ms);
// accumulate known signals
++total;
cancellostrace += signals == 1010111;
cancelled += signals == 1010011;
if (total != cancellostrace + cancelled) {
// display the unrecognized signals recorded
REQUIRE(signals == -1);
}
if (total >= 100) {
// too long, abort and show the signals distribution
WARN(
"total " << total << ", cancel-lost-race " << cancellostrace
<< ", cancelled " << cancelled);
break;
}
if (cancellostrace > 4 && cancelled > 4) {
// yay all known outcomes were observed!
break;
}
// try again
continue;
}
}
time.join(); std::atomic<int> signals_{0};
} std::atomic<int> terminal_{0};
std::atomic<int> cancel_{0};
std::chrono::system_clock::time_point at_{mi::now(tnt_) + 100ms};
};
TEST_F(ConcurrentFlowManySender, EarlyCancellation) {
// this nightmare brought to you by ASAN stack-use-after-return.
cancellation_test(at_ - 50ms);
join();
EXPECT_THAT(signals_, Eq(1010011))
<< "expected that the starting, up.done and out.done signals are each recorded once";
} }
SCENARIO("flow many from", "[flow][sender][for_each]") { TEST_F(ConcurrentFlowManySender, LateCancellation) {
GIVEN("A flow many sender of 5 values") { // this nightmare brought to you by ASAN stack-use-after-return.
auto v = std::array<int, 5>{0, 1, 2, 3, 4}; cancellation_test(at_ + 50ms);
auto f = op::flow_from(v);
join();
EXPECT_THAT(signals_, Eq(1010111))
<< "expected that the starting, up.done and out.value signals are each recorded once";
}
TEST_F(ConcurrentFlowManySender, RacingCancellation) {
int total = 0;
int cancellostrace = 0; // 1010111
int cancelled = 0; // 1010011
for (;;) {
reset();
cancellation_test(at_);
// accumulate known signals
++total;
cancellostrace += signals_ == 1010111;
cancelled += signals_ == 1010011;
WHEN("for_each is applied") { EXPECT_THAT(total, Eq(cancellostrace + cancelled))
int actual = 0; << signals_ << " <- this set of signals is unrecognized";
f | op::for_each(mi::MAKE(receiver)([&](int){++actual;}));
THEN("all the values are sent once") { ASSERT_THAT(total, Lt(100))
REQUIRE(actual == 5); // too long, abort and show the signals distribution
} << "total " << total << ", cancel-lost-race " << cancellostrace
<< ", cancelled " << cancelled;
if (cancellostrace > 4 && cancelled > 4) {
// yay all known outcomes were observed!
break;
} }
// try again
continue;
} }
join();
}
TEST(FlowManySender, From) {
auto v = std::array<int, 5>{0, 1, 2, 3, 4};
auto f = op::flow_from(v);
int actual = 0;
f | op::for_each(mi::MAKE(receiver)([&](int) { ++actual; }));
EXPECT_THAT(actual, Eq(5)) << "expexcted that all the values are sent once";
} }
#endif
...@@ -29,7 +29,10 @@ using namespace std::literals; ...@@ -29,7 +29,10 @@ using namespace std::literals;
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#if 0 #include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using namespace testing;
#if __cpp_deduction_guides >= 201703 #if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_ #define MAKE(x) x MAKE_
...@@ -39,11 +42,10 @@ using namespace folly::pushmi::aliases; ...@@ -39,11 +42,10 @@ using namespace folly::pushmi::aliases;
#define MAKE(x) make_##x #define MAKE(x) make_##x
#endif #endif
SCENARIO("flow single immediate cancellation", "[flow][sender]") { class ImmediateFlowSingleSender : public Test {
int signals = 0; protected:
auto make_producer() {
GIVEN("A flow single sender") { return mi::MAKE(flow_single_sender)([&](auto out) {
auto f = mi::MAKE(flow_single_sender)([&](auto out) {
// boolean cancellation // boolean cancellation
bool stop = false; bool stop = false;
auto set_stop = [](auto& stop) { auto set_stop = [](auto& stop) {
...@@ -60,13 +62,13 @@ SCENARIO("flow single immediate cancellation", "[flow][sender]") { ...@@ -60,13 +62,13 @@ SCENARIO("flow single immediate cancellation", "[flow][sender]") {
}; };
auto up = mi::MAKE(receiver)( auto up = mi::MAKE(receiver)(
Data{std::move(tokens.second)}, Data{std::move(tokens.second)},
[&](auto& data, auto e) noexcept { [&](auto& data, auto) noexcept {
signals += 100000; signals_ += 100000;
auto both = lock_both(data.stopper); auto both = lock_both(data.stopper);
(*(both.first))(both.second); (*(both.first))(both.second);
}, },
mi::on_done([&](auto& data) { mi::on_done([&](auto& data) {
signals += 10000; signals_ += 10000;
auto both = lock_both(data.stopper); auto both = lock_both(data.stopper);
(*(both.first))(both.second); (*(both.first))(both.second);
})); }));
...@@ -83,53 +85,69 @@ SCENARIO("flow single immediate cancellation", "[flow][sender]") { ...@@ -83,53 +85,69 @@ SCENARIO("flow single immediate cancellation", "[flow][sender]") {
::mi::set_done(out); ::mi::set_done(out);
} }
}); });
}
WHEN("submit is applied and cancels the producer") { template <class F>
f | auto make_consumer(F f) {
op::submit( return mi::MAKE(flow_receiver)(
mi::on_value([&](int) { signals += 100; }), mi::on_value([&](int) { signals_ += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }), mi::on_error([&](auto) noexcept { signals_ += 1000; }),
mi::on_done([&]() { signals += 1; }), mi::on_done([&]() { signals_ += 1; }),
// immediately stop producer mi::on_starting([&, f](auto up) {
mi::on_starting([&](auto up) { signals_ += 10;
signals += 10; f(std::move(up));
::mi::set_done(up); }));
})); }
THEN( int signals_{0};
"the starting, up.done and out.done signals are each recorded once") { };
REQUIRE(signals == 10011);
}
}
WHEN("submit is applied and cancels the producer late") { TEST_F(ImmediateFlowSingleSender, EarlyCancellation) {
f | make_producer() | op::submit(make_consumer([](auto up) {
op::submit( // immediately stop producer
mi::on_value([&](int) { signals += 100; }), ::mi::set_done(up);
mi::on_error([&](auto) noexcept { signals += 1000; }), }));
mi::on_done([&]() { signals += 1; }),
// do not stop producer before it is scheduled to run EXPECT_THAT(signals_, Eq(10011))
mi::on_starting([&](auto up) { signals += 10; })); << "expected that the starting, up.done and out.done signals are each recorded once";
}
THEN(
"the starting and out.value signals are each recorded once") { TEST_F(ImmediateFlowSingleSender, LateCancellation) {
REQUIRE(signals == 110); make_producer() | op::submit(make_consumer([](auto) {
} // do not stop producer before it is scheduled to run
} }));
}
EXPECT_THAT(signals_, Eq(110))
<< "expected that the starting and out.value signals are each recorded once";
} }
SCENARIO("flow single shared cancellation new thread", "[flow][sender]") { using NT = decltype(mi::new_thread());
auto nt = mi::new_thread();
using NT = decltype(nt); inline auto make_time(mi::time_source<>& t, NT& ex) {
auto time = mi::time_source<>{}; return t.make(mi::systemNowF{}, [ex]() { return ex; })();
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })(); }
auto tcncl = time.make(mi::systemNowF{}, [nt](){ return nt; })();
std::atomic<int> signals{0}; class ConcurrentFlowSingleSender : public Test {
auto at = mi::now(tnt) + 200ms; protected:
using TNT = mi::invoke_result_t<decltype(make_time), mi::time_source<>&, NT&>;
GIVEN("A flow single sender") {
auto f = mi::MAKE(flow_single_sender)([&](auto out) { void reset() {
at_ = mi::now(tnt_) + 100ms;
signals_ = 0;
terminal_ = 0;
cancel_ = 0;
}
void join() {
timeproduce_.join();
timecancel_.join();
}
template <class MakeTokens>
void cancellation_test(
std::chrono::system_clock::time_point at,
MakeTokens make_tokens) {
auto f = mi::MAKE(flow_single_sender)([&, make_tokens](auto out) {
// boolean cancellation // boolean cancellation
bool stop = false; bool stop = false;
auto set_stop = [](auto& stop) { auto set_stop = [](auto& stop) {
...@@ -137,7 +155,7 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") { ...@@ -137,7 +155,7 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") {
*stop = true; *stop = true;
} }
}; };
auto tokens = mi::shared_entangle(stop, set_stop); auto tokens = make_tokens(stop, set_stop);
using Stopper = decltype(tokens.second); using Stopper = decltype(tokens.second);
struct Data : mi::receiver<> { struct Data : mi::receiver<> {
...@@ -146,30 +164,32 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") { ...@@ -146,30 +164,32 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") {
}; };
auto up = mi::MAKE(receiver)( auto up = mi::MAKE(receiver)(
Data{std::move(tokens.second)}, Data{std::move(tokens.second)},
[&](auto& data, auto e) noexcept { [&](auto& data, auto) noexcept {
signals += 100000; signals_ += 100000;
++cancel_;
auto both = lock_both(data.stopper); auto both = lock_both(data.stopper);
(*(both.first))(both.second); (*(both.first))(both.second);
}, },
mi::on_done([&](auto& data) { mi::on_done([&](auto& data) {
signals += 10000; signals_ += 10000;
++cancel_;
auto both = lock_both(data.stopper); auto both = lock_both(data.stopper);
(*(both.first))(both.second); (*(both.first))(both.second);
})); }));
// make all the signals come from the same thread // make all the signals come from the same thread
tnt | tnt_ |
op::submit([stoppee = std::move(tokens.first), op::submit([&,
up = std::move(up), stoppee = std::move(tokens.first),
out = std::move(out), up_not_a_shadow_howtoeven = std::move(up),
at](auto tnt) mutable { out = std::move(out)](auto tnt) mutable {
// pass reference for cancellation. // pass reference for cancellation.
::mi::set_starting(out, std::move(up)); ::mi::set_starting(out, std::move(up_not_a_shadow_howtoeven));
// submit work to happen later // submit work to happen later
tnt | tnt |
op::submit_at( op::submit_at(
at, at_,
[stoppee = std::move(stoppee), [stoppee = std::move(stoppee),
out = std::move(out)](auto) mutable { out = std::move(out)](auto) mutable {
// check boolean to select signal // check boolean to select signal
...@@ -185,289 +205,152 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") { ...@@ -185,289 +205,152 @@ SCENARIO("flow single shared cancellation new thread", "[flow][sender]") {
}); });
}); });
WHEN("submit is applied and cancels the producer early") { f |
{ op::submit(
f | mi::on_value([&](int) { signals_ += 100; }),
op::blocking_submit( mi::on_error([&](auto) noexcept {
mi::on_value([&](int) { signals += 100; }), signals_ += 1000;
mi::on_error([&](auto) noexcept { signals += 1000; }), ++terminal_;
mi::on_done([&]() { signals += 1; }), }),
// stop producer before it is scheduled to run mi::on_done([&]() {
mi::on_starting([&](auto up) { signals_ += 1;
signals += 10; ++terminal_;
tcncl | }),
op::submit_at( // stop producer before it is scheduled to run
at - 50ms, [up = std::move(up)](auto) mutable { mi::on_starting([&, at](auto up) {
::mi::set_done(up); signals_ += 10;
}); tcncl_ | op::submit_at(at, [up = std::move(up)](auto) mutable {
})); ::mi::set_done(up);
} });
}));
// make sure that the completion signal arrives
std::this_thread::sleep_for(100ms);
THEN( while (terminal_ == 0 || cancel_ == 0) {
"the starting, up.done and out.done signals are each recorded once") { std::this_thread::yield();
REQUIRE(signals == 10011);
}
} }
}
WHEN("submit is applied and cancels the producer late") { NT ntproduce_{mi::new_thread()};
{ mi::time_source<> timeproduce_{};
f | TNT tnt_{make_time(timeproduce_, ntproduce_)};
op::blocking_submit(
mi::on_value([&](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;
tcncl |
op::submit_at(
at + 50ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
std::this_thread::sleep_for(100ms); NT ntcancel_{mi::new_thread()};
mi::time_source<> timecancel_{};
TNT tcncl_{make_time(timecancel_, ntcancel_)};
THEN( std::atomic<int> signals_{0};
"the starting, up.done, out.value and out.done signals are each recorded once") { std::atomic<int> terminal_{0};
REQUIRE(signals == 10111); std::atomic<int> cancel_{0};
} std::chrono::system_clock::time_point at_{mi::now(tnt_) + 100ms};
} };
WHEN("submit is applied and cancels the producer at the same time") { TEST_F(ConcurrentFlowSingleSender, EarlySharedCancellation) {
// count known results cancellation_test(at_ - 50ms, [](auto& stop, auto& set_stop) {
int total = 0; return mi::shared_entangle(stop, set_stop);
int cancellostrace = 0; // 10111 });
int cancelled = 0; // 10011
for (;;) {
signals = 0;
// set completion time to be in 100ms
at = mi::now(tnt) + 100ms;
{
f |
op::blocking_submit(
mi::on_value([&](int) { signals += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }),
mi::on_done([&]() { signals += 1; }),
// stop producer at the same time that it is scheduled to run
mi::on_starting([&](auto up) {
signals += 10;
tcncl | op::submit_at(at, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
// make sure any cancellation signal has completed
std::this_thread::sleep_for(100ms);
// accumulate known signals join();
++total;
cancellostrace += signals == 10111;
cancelled += signals == 10011;
if (total != cancellostrace + cancelled) { EXPECT_THAT(signals_, Eq(10011));
// display the unrecognized signals recorded
REQUIRE(signals == -1);
}
if (total >= 100) {
// too long, abort and show the signals distribution
WARN(
"total " << total << ", cancel-lost-race " << cancellostrace
<< ", cancelled " << cancelled);
break;
}
if (cancellostrace > 4 && cancelled > 4) {
// yay all known outcomes were observed!
break;
}
// try again
continue;
}
}
time.join();
}
} }
SCENARIO("flow single entangled cancellation new thread", "[flow][sender]") { TEST_F(ConcurrentFlowSingleSender, LateSharedCancellation) {
auto nt = mi::new_thread(); cancellation_test(at_ + 50ms, [](auto& stop, auto& set_stop) {
using NT = decltype(nt); return mi::shared_entangle(stop, set_stop);
auto time = mi::time_source<>{}; });
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })();
auto tcncl = time.make(mi::systemNowF{}, [nt](){ return nt; })();
std::atomic<int> signals{0};
auto at = mi::now(tnt) + 200ms;
GIVEN("A flow single sender") {
auto f = mi::MAKE(flow_single_sender)([&](auto out) {
// boolean cancellation
bool stop = false;
auto set_stop = [](auto& stop) {
if (!!stop) {
*stop = true;
}
};
auto tokens = mi::entangle(stop, set_stop);
using Stopper = decltype(tokens.second); join();
struct Data : mi::receiver<> {
explicit Data(Stopper stopper) : stopper(std::move(stopper)) {}
Stopper stopper;
};
auto up = mi::MAKE(receiver)(
Data{std::move(tokens.second)},
[&](auto& data, auto e) noexcept {
signals += 100000;
auto both = lock_both(data.stopper);
(*(both.first))(both.second);
},
mi::on_done([&](auto& data) {
signals += 10000;
auto both = lock_both(data.stopper);
(*(both.first))(both.second);
}));
// make all the signals come from the same thread EXPECT_THAT(signals_, Eq(10111))
tnt | << "expected that the starting, up.done, out.value and out.done signals are each recorded once";
op::submit([stoppee = std::move(tokens.first), }
up = std::move(up),
out = std::move(out),
at](auto tnt) mutable {
// pass reference for cancellation.
::mi::set_starting(out, std::move(up));
// submit work to happen later TEST_F(ConcurrentFlowSingleSender, RacingSharedCancellation) {
tnt | int total = 0;
op::submit_at( int cancellostrace = 0; // 10111
at, int cancelled = 0; // 10011
[stoppee = std::move(stoppee),
out = std::move(out)](auto) mutable { for (;;) {
// check boolean to select signal reset();
auto both = lock_both(stoppee); cancellation_test(at_, [](auto& stop, auto& set_stop) {
if (!!both.first && !*(both.first)) { return mi::shared_entangle(stop, set_stop);
::mi::set_value(out, 42);
::mi::set_done(out);
} else {
// cancellation is not an error
::mi::set_done(out);
}
});
});
}); });
WHEN("submit is applied and cancels the producer early") { // accumulate known signals
{ ++total;
f | cancellostrace += signals_ == 10111;
op::blocking_submit( cancelled += signals_ == 10011;
mi::on_value([&](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;
tcncl |
op::submit_at(
at - 50ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
// make sure that the completion signal arrives EXPECT_THAT(total, Eq(cancellostrace + cancelled))
std::this_thread::sleep_for(100ms); << signals_ << " <- this set of signals is unrecognized";
THEN( ASSERT_THAT(total, Lt(100))
"the starting, up.done and out.done signals are each recorded once") { // too long, show the signals distribution
REQUIRE(signals == 10011); << "total " << total << ", cancel-lost-race " << cancellostrace
} << ", cancelled " << cancelled;
if (cancellostrace > 4 && cancelled > 4) {
// yay all known outcomes were observed!
break;
} }
// try again
continue;
}
WHEN("submit is applied and cancels the producer late") { join();
{ }
f |
op::blocking_submit(
mi::on_value([&](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;
tcncl |
op::submit_at(
at + 50ms, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
std::this_thread::sleep_for(100ms); TEST_F(ConcurrentFlowSingleSender, EarlyEntangledCancellation) {
cancellation_test(at_ - 50ms, [](auto& stop, auto& set_stop) {
return mi::entangle(stop, set_stop);
});
THEN( join();
"the starting, up.done, out.value and out.done signals are each recorded once") {
REQUIRE(signals == 10111);
}
}
WHEN("submit is applied and cancels the producer at the same time") { EXPECT_THAT(signals_, Eq(10011));
// count known results }
int total = 0;
int cancellostrace = 0; // 10111
int cancelled = 0; // 10011
for (;;) {
signals = 0;
// set completion time to be in 100ms
at = mi::now(tnt) + 100ms;
{
f |
op::blocking_submit(
mi::on_value([&](int) { signals += 100; }),
mi::on_error([&](auto) noexcept { signals += 1000; }),
mi::on_done([&]() { signals += 1; }),
// stop producer at the same time that it is scheduled to run
mi::on_starting([&](auto up) {
signals += 10;
tcncl | op::submit_at(at, [up = std::move(up)](auto) mutable {
::mi::set_done(up);
});
}));
}
// make sure any cancellation signal has completed TEST_F(ConcurrentFlowSingleSender, LateEntangledCancellation) {
std::this_thread::sleep_for(200ms); cancellation_test(at_ + 50ms, [](auto& stop, auto& set_stop) {
return mi::entangle(stop, set_stop);
});
// accumulate known signals join();
++total;
cancellostrace += signals == 10111;
cancelled += signals == 10011;
if (total != cancellostrace + cancelled) { EXPECT_THAT(signals_, Eq(10111))
// display the unrecognized signals recorded << "expected that the starting, up.done, out.value and out.done signals are each recorded once";
REQUIRE(signals == -1); }
}
if (total >= 100) { TEST_F(ConcurrentFlowSingleSender, RacingEntangledCancellation) {
// too long, abort and show the signals distribution int total = 0;
WARN( int cancellostrace = 0; // 10111
"total " << total << ", cancel-lost-race " << cancellostrace int cancelled = 0; // 10011
<< ", cancelled " << cancelled);
break; for (;;) {
} reset();
if (cancellostrace > 4 && cancelled > 4) { cancellation_test(at_, [](auto& stop, auto& set_stop) {
// yay all known outcomes were observed! return mi::entangle(stop, set_stop);
break; });
}
// try again
continue;
}
}
time.join(); // accumulate known signals
++total;
cancellostrace += signals_ == 10111;
cancelled += signals_ == 10011;
EXPECT_THAT(total, Eq(cancellostrace + cancelled))
<< signals_ << " <- this set of signals is unrecognized";
ASSERT_THAT(total, Lt(100))
// too long, show the signals distribution
<< "total " << total << ", cancel-lost-race " << cancellostrace
<< ", cancelled " << cancelled;
if (cancellostrace > 4 && cancelled > 4) {
// yay all known outcomes were observed!
break;
}
// try again
continue;
} }
join();
} }
#endif
...@@ -21,24 +21,29 @@ using namespace std::literals; ...@@ -21,24 +21,29 @@ using namespace std::literals;
#include <folly/experimental/pushmi/flow_single_sender.h> #include <folly/experimental/pushmi/flow_single_sender.h>
#include <folly/experimental/pushmi/o/empty.h> #include <folly/experimental/pushmi/o/empty.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/just.h> #include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/on.h> #include <folly/experimental/pushmi/o/on.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/tap.h> #include <folly/experimental/pushmi/o/tap.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h> #include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/new_thread.h> #include <folly/experimental/pushmi/new_thread.h>
#include <folly/experimental/pushmi/time_source.h>
#include <folly/experimental/pushmi/strand.h> #include <folly/experimental/pushmi/strand.h>
#include <folly/experimental/pushmi/time_source.h>
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#if 0 #include <folly/Conv.h>
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using namespace testing;
struct countdownsingle { struct countdownsingle {
countdownsingle(int& c) explicit countdownsingle(int& c) : counter(&c) {}
: counter(&c) {}
int* counter; int* counter;
...@@ -50,173 +55,167 @@ struct countdownsingle { ...@@ -50,173 +55,167 @@ struct countdownsingle {
} }
}; };
SCENARIO( "new_thread executor", "[new_thread][sender]" ) { using NT = decltype(mi::new_thread());
GIVEN( "A new_thread time_single_sender" ) { inline auto make_time(mi::time_source<>& t, NT& ex) {
auto nt = v::new_thread(); return t.make(mi::systemNowF{}, [ex]() { return ex; })();
using NT = decltype(nt); }
auto time = mi::time_source<>{}; class NewthreadExecutor : public Test {
public:
~NewthreadExecutor() override {
time_.join();
}
auto tnt = time.make(mi::systemNowF{}, [nt](){ return nt; })(); protected:
using TNT = mi::invoke_result_t<decltype(make_time), mi::time_source<>&, NT&>;
WHEN( "blocking submit now" ) { NT nt_{mi::new_thread()};
auto signals = 0; mi::time_source<> time_{};
auto start = v::now(tnt); TNT tnt_{make_time(time_, nt_)};
auto signaled = start; };
tnt |
op::transform([](auto tnt){ return tnt | ep::now(); }) | TEST_F(NewthreadExecutor, BlockingSubmitNow) {
op::blocking_submit( auto signals = 0;
[&](auto at){ auto start = v::now(tnt_);
auto signaled = start;
tnt_ | op::transform([](auto tnt) { return tnt | ep::now(); }) |
op::blocking_submit(
[&](auto at) {
signaled = at; signaled = at;
signals += 100; }, signals += 100;
[&](auto e) noexcept { signals += 1000; }, },
[&](){ signals += 10; }); [&](auto) noexcept { signals += 1000; },
[&]() { signals += 10; });
THEN( "the value and done signals are recorded once and the value signal did not drift much" ) {
REQUIRE( signals == 110 ); EXPECT_THAT(signals, Eq(110))
auto delay = std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start)).count(); << "expected that the value and done signals are recorded once and the value signal did not drift much";
INFO("The delay is " << ::Catch::Detail::stringify(delay)); auto delay =
REQUIRE( delay < 1000 ); std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start))
} .count();
} EXPECT_THAT(delay, Lt(1000)) << "The delay is " << delay;
}
WHEN( "blocking get now" ) { TEST_F(NewthreadExecutor, BlockingGetNow) {
auto start = v::now(tnt); auto start = v::now(tnt_);
auto signaled = tnt | auto signaled = tnt_ | op::transform([](auto tnt) { return v::now(tnt); }) |
op::transform([](auto tnt){ op::get<std::chrono::system_clock::time_point>;
return v::now(tnt);
}) |
op::get<std::chrono::system_clock::time_point>;
THEN( "the signal did not drift much" ) {
auto delay = std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start)).count();
INFO("The delay is " << ::Catch::Detail::stringify(delay));
REQUIRE( delay < 1000 );
}
}
WHEN( "submissions are ordered in time" ) { auto delay =
std::vector<std::string> times; std::chrono::duration_cast<std::chrono::milliseconds>((signaled - start))
std::atomic<int> pushed(0); .count();
auto push = [&](int time) {
return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); ++pushed; });
};
tnt | op::submit(v::on_value([push](auto tnt) {
auto now = tnt | ep::now();
tnt |
op::submit_after(40ms, push(40)) |
op::submit_at(now + 10ms, push(10)) |
op::submit_after(20ms, push(20)) |
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" ) {
REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"});
}
}
WHEN( "now is called" ) { EXPECT_THAT(delay, Lt(1000)) << "The delay is " << delay;
bool done = false; }
tnt | ep::now();
tnt | op::blocking_submit([&](auto tnt) {
tnt | ep::now();
done = true;
});
THEN( "both calls to now() complete" ) {
REQUIRE( done == true );
}
}
WHEN( "blocking submit" ) { TEST_F(NewthreadExecutor, SubmissionsAreOrderedInTime) {
auto signals = 0; std::vector<std::string> times;
nt | std::atomic<int> pushed{0};
op::transform([](auto){ return 42; }) | auto push = [&](int time) {
op::blocking_submit( return v::on_value([&, time](auto) {
[&](auto){ times.push_back(folly::to<std::string>(time));
signals += 100; }, ++pushed;
[&](auto e) noexcept { signals += 1000; }, });
[&](){ signals += 10; }); };
tnt_ | op::submit(v::on_value([push](auto tnt) {
THEN( "the value and done signals are recorded once" ) { auto now = tnt | ep::now();
REQUIRE( signals == 110 ); tnt | op::submit_after(40ms, push(40)) |
} op::submit_at(now + 10ms, push(10)) | op::submit_after(20ms, push(20)) |
} op::submit_at(now + 10ms, push(11));
}));
while (pushed.load() < 4) {
std::this_thread::yield();
}
WHEN( "blocking get" ) { EXPECT_THAT(times, ElementsAre("10", "11", "20", "40"))
auto v = nt | << "expected that the items were pushed in time order not insertion order";
op::transform([](auto){ }
return 42;
}) |
op::get<int>;
THEN( "the result is" ) { TEST_F(NewthreadExecutor, NowIsCalled) {
REQUIRE( v == 42 ); bool done = false;
} tnt_ | ep::now();
} tnt_ | op::blocking_submit([&](auto tnt) {
tnt | ep::now();
done = true;
});
WHEN( "virtual derecursion is triggered" ) { EXPECT_THAT(done, Eq(true)) << "exptected that both calls to now() complete";
int counter = 100'000; }
std::function<void(::folly::pushmi::any_executor_ref<> exec)> recurse;
recurse = [&](::folly::pushmi::any_executor_ref<> nt) {
if (--counter <= 0)
return;
nt | op::submit(recurse);
};
nt | op::blocking_submit([&](auto nt) { recurse(nt); });
THEN( "all nested submissions complete" ) {
REQUIRE( counter == 0 );
}
}
WHEN( "static derecursion is triggered" ) { TEST_F(NewthreadExecutor, BlockingSubmit) {
int counter = 100'000; auto signals = 0;
countdownsingle single{counter}; nt_ | op::transform([](auto) { return 42; }) |
nt | op::blocking_submit(single); op::blocking_submit(
THEN( "all nested submissions complete" ) { [&](auto) { signals += 100; },
REQUIRE( counter == 0 ); [&](auto) noexcept { signals += 1000; },
} [&]() { signals += 10; });
}
WHEN( "used with on" ) { EXPECT_THAT(signals, Eq(110))
std::vector<std::string> values; << "the value and done signals are recorded once";
auto sender = ::folly::pushmi::make_single_sender([](auto out) { }
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
sender | op::on([&](){return nt;}) |
op::blocking_submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); }));
THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"});
}
}
WHEN( "used with via" ) { TEST_F(NewthreadExecutor, BlockingGet) {
std::vector<std::string> values; auto v = nt_ | op::transform([](auto) { return 42; }) | op::get<int>;
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
sender | op::via(mi::strands(nt)) |
op::blocking_submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); }));
THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"});
}
}
time.join(); EXPECT_THAT(v, Eq(42)) << "expected that the result would be different";
} }
TEST_F(NewthreadExecutor, VirtualDerecursion) {
int counter = 100'000;
std::function<void(::folly::pushmi::any_executor_ref<> exec)> recurse;
recurse = [&](::folly::pushmi::any_executor_ref<> nt) {
if (--counter <= 0)
return;
nt | op::submit(recurse);
};
nt_ | op::blocking_submit([&](auto nt) { recurse(nt); });
EXPECT_THAT(counter, Eq(0))
<< "expected that all nested submissions complete";
}
TEST_F(NewthreadExecutor, StaticDerecursion) {
int counter = 100'000;
countdownsingle single{counter};
nt_ | op::blocking_submit(single);
EXPECT_THAT(counter, Eq(0))
<< "expected that all nested submissions complete";
}
TEST_F(NewthreadExecutor, UsedWithOn) {
std::vector<std::string> values;
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
sender | op::on([&]() { return nt_; }) |
op::blocking_submit(v::on_value(
[&](auto v) { values.push_back(folly::to<std::string>(v)); }));
EXPECT_THAT(values, ElementsAre(folly::to<std::string>(2.0)))
<< "expected that only the first item was pushed";
}
TEST_F(NewthreadExecutor, UsedWithVia) {
std::vector<std::string> values;
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
sender | op::via(mi::strands(nt_)) |
op::blocking_submit(v::on_value(
[&](auto v) { values.push_back(folly::to<std::string>(v)); }));
EXPECT_THAT(values, ElementsAre(folly::to<std::string>(2.0)))
<< "expected that only the first item was pushed";
} }
#endif
...@@ -29,140 +29,169 @@ using namespace std::literals; ...@@ -29,140 +29,169 @@ using namespace std::literals;
#include <folly/experimental/pushmi/o/tap.h> #include <folly/experimental/pushmi/o/tap.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/trampoline.h>
#include <folly/experimental/pushmi/new_thread.h> #include <folly/experimental/pushmi/new_thread.h>
#include <folly/experimental/pushmi/trampoline.h>
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#if 0 #include <folly/portability/GMock.h>
SCENARIO( "empty can be used with tap and submit", "[empty][sender]" ) { #include <folly/portability/GTest.h>
GIVEN( "An empty sender" ) { using namespace testing;
auto e = op::empty();
using E = decltype(e); TEST(EmptyNoArgSingleSender, TapAndSubmit) {
auto e = op::empty();
REQUIRE( v::SenderTo<E, v::any_receiver<>, v::is_single<>> ); using E = decltype(e);
WHEN( "tap and submit are applied" ) { EXPECT_THAT((v::SenderTo<E, v::any_receiver<>, v::is_single<>>), Eq(true))
int signals = 0; << "expected empty to return a single sender that can take an any_receiver";
e |
op::tap( int signals = 0;
[&](){ signals += 100; }, e |
[&](auto e) noexcept { signals += 1000; }, op::tap(
[&](){ signals += 10; }) | [&]() { signals += 100; },
op::submit( [&](auto) noexcept { signals += 1000; },
[&](){ signals += 100; }, [&]() { signals += 10; }) |
[&](auto e) noexcept { signals += 1000; }, op::submit(
[&](){ signals += 10; }); [&]() { signals += 100; },
[&](auto) noexcept { signals += 1000; },
THEN( "the done signal is recorded twice" ) { [&]() { signals += 10; });
REQUIRE( signals == 20 );
} EXPECT_THAT(signals, Eq(20))
<< "expected the done signal to be recorded twice";
WHEN( "future_from is applied" ) {
REQUIRE_THROWS_AS(v::future_from(e).get(), std::future_error); EXPECT_THROW(v::future_from(e).get(), std::future_error)
<< "expected future_error when future_from is applied";
THEN( "future_from(e) returns std::future<void>" ) {
REQUIRE( std::is_same<std::future<void>, decltype(v::future_from(e))>::value ); EXPECT_THAT(
} (std::is_same<std::future<void>, decltype(v::future_from(e))>::value),
} Eq(true))
} << "expected future_from(e) to return std::future<void>";
} }
GIVEN( "An empty int single_sender" ) { TEST(EmptyIntSingleSender, TapAndSubmit) {
auto e = op::empty<int>(); auto e = op::empty<int>();
using E = decltype(e); using E = decltype(e);
REQUIRE( v::SenderTo<E, v::any_receiver<std::exception_ptr, int>, v::is_single<>> ); EXPECT_THAT(
(v::SenderTo<
WHEN( "tap and submit are applied" ) { E,
v::any_receiver<std::exception_ptr, int>,
int signals = 0; v::is_single<>>),
e | Eq(true))
op::tap( << "expected empty to return a single sender that can take an any_receiver<int>";
[&](auto v){ signals += 100; },
[&](auto e) noexcept { signals += 1000; }, int signals = 0;
[&](){ signals += 10; }) | e |
op::submit( op::tap(
[&](auto v){ signals += 100; }, [&](auto) { signals += 100; },
[&](auto e) noexcept { signals += 1000; }, [&](auto) noexcept { signals += 1000; },
[&](){ signals += 10; }); [&]() { signals += 10; }) |
op::submit(
THEN( "the done signal is recorded twice" ) { [&](auto) { signals += 100; },
REQUIRE( signals == 20 ); [&](auto) noexcept { signals += 1000; },
} [&]() { signals += 10; });
}
} EXPECT_THAT(signals, Eq(20))
<< "expected the done signal to be recorded twice";
EXPECT_THROW(v::future_from<int>(e).get(), std::future_error)
<< "expected future_error when future_from is applied";
EXPECT_THAT(
(std::is_same<std::future<int>, decltype(v::future_from<int>(e))>::value),
Eq(true))
<< "expected future_from(e) to return std::future<void>";
} }
SCENARIO( "just() can be used with transform and submit", "[just][sender]" ) { TEST(JustIntSingleSender, TransformAndSubmit) {
auto j = op::just(20);
GIVEN( "A just int single_sender" ) { using J = decltype(j);
auto j = op::just(20);
using J = decltype(j); EXPECT_THAT(
(v::SenderTo<
REQUIRE( v::SenderTo<J, v::any_receiver<std::exception_ptr, int>, v::is_single<>> ); J,
v::any_receiver<std::exception_ptr, int>,
WHEN( "transform and submit are applied" ) { v::is_single<>>),
int signals = 0; Eq(true))
int value = 0; << "expected empty to return a single sender that can take an any_receiver<int>";
j |
op::transform( int signals = 0;
[&](int v){ signals += 10000; return v + 1; }, int value = 0;
[&](auto v){ std:abort(); return v; }) | j |
op::transform( op::transform(
[&](int v){ signals += 10000; return v * 2; }) | [&](int v) {
op::submit( signals += 10000;
[&](auto v){ value = v; signals += 100; }, return v + 1;
[&](auto e) noexcept { signals += 1000; }, },
[&](){ signals += 10; }); [&](auto v) {
std::abort();
THEN( "the transform signal is recorded twice, the value and done signals once and the result is correct" ) { return v;
REQUIRE( signals == 20110 ); }) |
REQUIRE( value == 42 ); op::transform([&](int v) {
} signals += 10000;
} return v * 2;
}) |
WHEN( "future_from<int> is applied" ) { op::submit(
auto twenty = v::future_from<int>(j).get(); [&](auto v) {
value = v;
THEN( "the value signal is recorded once and the result is correct" ) { signals += 100;
REQUIRE( twenty == 20 ); },
REQUIRE( std::is_same<std::future<int>, decltype(v::future_from<int>(j))>::value ); [&](auto) noexcept { signals += 1000; },
} [&]() { signals += 10; });
}
} EXPECT_THAT(signals, Eq(20110))
<< "expected that the transform signal is recorded twice and that the value and done signals once each";
EXPECT_THAT(value, Eq(42)) << "expected a different result";
auto twenty = v::future_from<int>(j).get();
EXPECT_THAT(twenty, Eq(20))
<< "expected a different result from future_from(e).get()";
EXPECT_THAT(
(std::is_same<std::future<int>, decltype(v::future_from<int>(j))>::value),
Eq(true))
<< "expected future_from(e) to return std::future<int>";
} }
SCENARIO( "from() can be used with transform and submit", "[from][sender]" ) { TEST(FromIntManySender, TransformAndSubmit) {
std::array<int, 3> arr{0, 9, 99};
GIVEN( "A from int many_sender" ) { auto m = op::from(arr);
int arr[] = {0, 9, 99}; using M = decltype(m);
auto m = op::from(arr);
using M = decltype(m); EXPECT_THAT(
(v::SenderTo<M, v::any_receiver<std::exception_ptr, int>, v::is_many<>>),
REQUIRE( v::SenderTo<M, v::any_receiver<std::exception_ptr, int>, v::is_many<>> ); Eq(true))
<< "expected empty to return a many sender that can take an any_receiver<int>";
WHEN( "transform and submit are applied" ) {
int signals = 0; int signals = 0;
int value = 0; int value = 0;
m | m |
op::transform( op::transform(
[&](int v){ signals += 10000; return v + 1; }, [&](int v) {
[&](auto v){ std:abort(); return v; }) | signals += 10000;
op::transform( return v + 1;
[&](int v){ signals += 10000; return v * 2; }) | },
op::submit( [&](auto v) {
[&](auto v){ value += v; signals += 100; }, std::abort();
[&](auto e) noexcept { signals += 1000; }, return v;
[&](){ signals += 10; }); }) |
op::transform([&](int v) {
THEN( "the transform signal is recorded twice, the value signal once and the result is correct" ) { signals += 10000;
REQUIRE( signals == 60310 ); return v * 2;
REQUIRE( value == 222 ); }) |
} op::submit(
} [&](auto v) {
value += v;
} signals += 100;
},
[&](auto) noexcept { signals += 1000; },
[&]() { signals += 10; });
EXPECT_THAT(signals, Eq(60310))
<< "expected that the transform signal is recorded six times and that the value signal three times and done signal once";
EXPECT_THAT(value, Eq(222)) << "expected a different result";
} }
#endif
...@@ -21,23 +21,28 @@ using namespace std::literals; ...@@ -21,23 +21,28 @@ using namespace std::literals;
#include <folly/experimental/pushmi/flow_single_sender.h> #include <folly/experimental/pushmi/flow_single_sender.h>
#include <folly/experimental/pushmi/o/empty.h> #include <folly/experimental/pushmi/o/empty.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/just.h> #include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/on.h> #include <folly/experimental/pushmi/o/on.h>
#include <folly/experimental/pushmi/o/transform.h> #include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/tap.h> #include <folly/experimental/pushmi/o/tap.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h> #include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/inline.h> #include <folly/experimental/pushmi/inline.h>
#include <folly/experimental/pushmi/trampoline.h> #include <folly/experimental/pushmi/trampoline.h>
using namespace folly::pushmi::aliases; using namespace folly::pushmi::aliases;
#if 0 #include <folly/Conv.h>
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using namespace testing;
struct countdownsingle { struct countdownsingle {
countdownsingle(int& c) explicit countdownsingle(int& c) : counter(&c) {}
: counter(&c) {}
int* counter; int* counter;
...@@ -49,102 +54,102 @@ struct countdownsingle { ...@@ -49,102 +54,102 @@ struct countdownsingle {
} }
}; };
SCENARIO( "trampoline executor", "[trampoline][sender]" ) { using TR = decltype(mi::trampoline());
GIVEN( "A trampoline single_sender" ) {
auto tr = v::trampoline();
using TR = decltype(tr);
WHEN( "submit" ) {
auto signals = 0;
tr |
op::transform([](auto){ return 42; }) |
op::submit(
[&](auto){
signals += 100; },
[&](auto e) noexcept { signals += 1000; },
[&](){ signals += 10; });
THEN( "the value and done signals are each recorded once" ) {
REQUIRE( signals == 110 );
}
}
WHEN( "blocking get" ) { class TrampolineExecutor : public Test {
auto v = tr | protected:
op::transform([](auto){ return 42; }) | TR tr_{mi::trampoline()};
op::get<int>; };
THEN( "the result is" ) { TEST_F(TrampolineExecutor, TransformAndSubmit) {
REQUIRE( v == 42 ); auto signals = 0;
} tr_ | op::transform([](auto) { return 42; }) |
} op::submit(
[&](auto) { signals += 100; },
[&](auto) noexcept { signals += 1000; },
[&]() { signals += 10; });
WHEN( "virtual derecursion is triggered" ) { EXPECT_THAT(signals, Eq(110))
int counter = 100'000; << "expected that the value and done signals are each recorded once";
std::function<void(::folly::pushmi::any_executor_ref<> exec)> recurse; }
recurse = [&](::folly::pushmi::any_executor_ref<> tr) {
if (--counter <= 0)
return;
tr | op::submit(recurse);
};
tr | op::submit([&](auto exec) { recurse(exec); });
THEN( "all nested submissions complete" ) {
REQUIRE( counter == 0 );
}
}
WHEN( "static derecursion is triggered" ) { TEST_F(TrampolineExecutor, BlockingGet) {
int counter = 100'000; auto v = tr_ | op::transform([](auto) { return 42; }) | op::get<int>;
countdownsingle single{counter};
tr | op::submit(single);
THEN( "all nested submissions complete" ) {
REQUIRE( counter == 0 );
}
}
WHEN( "used with on" ) { EXPECT_THAT(v, Eq(42)) << "expected that the result would be different";
std::vector<std::string> values; }
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
auto inlineon = sender | op::on([&](){return mi::inline_executor();});
inlineon |
op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); }));
THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"});
}
THEN( "executor was not changed by on" ) {
REQUIRE(std::is_same<mi::executor_t<decltype(sender)>, mi::executor_t<decltype(inlineon)>>::value);
}
}
WHEN( "used with via" ) { TEST_F(TrampolineExecutor, VirtualDerecursion) {
std::vector<std::string> values; int counter = 100'000;
auto sender = ::folly::pushmi::make_single_sender([](auto out) { std::function<void(::folly::pushmi::any_executor_ref<> exec)> recurse;
::folly::pushmi::set_value(out, 2.0); recurse = [&](::folly::pushmi::any_executor_ref<> tr) {
::folly::pushmi::set_done(out); if (--counter <= 0)
// ignored return;
::folly::pushmi::set_value(out, 1); tr | op::submit(recurse);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min()); };
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max()); tr_ | op::submit([&](auto exec) { recurse(exec); });
});
auto inlinevia = sender | op::via([&](){return mi::inline_executor();}); EXPECT_THAT(counter, Eq(0))
inlinevia | << "expected that all nested submissions complete";
op::submit(v::on_value([&](auto v) { values.push_back(std::to_string(v)); })); }
THEN( "only the first item was pushed" ) {
REQUIRE(values == std::vector<std::string>{"2.000000"}); TEST_F(TrampolineExecutor, StaticDerecursion) {
} int counter = 100'000;
THEN( "executor was changed by via" ) { countdownsingle single{counter};
REQUIRE(!std::is_same<mi::executor_t<decltype(sender)>, mi::executor_t<decltype(inlinevia)>>::value); tr_ | op::submit(single);
}
} EXPECT_THAT(counter, Eq(0))
} << "expected that all nested submissions complete";
}
TEST_F(TrampolineExecutor, UsedWithOn) {
std::vector<std::string> values;
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
auto inlineon = sender | op::on([&]() { return mi::inline_executor(); });
inlineon | op::submit(v::on_value([&](auto v) {
values.push_back(folly::to<std::string>(v));
}));
EXPECT_THAT(values, ElementsAre(folly::to<std::string>(2.0)))
<< "expected that only the first item was pushed";
EXPECT_THAT(
(std::is_same<
mi::executor_t<decltype(sender)>,
mi::executor_t<decltype(inlineon)>>::value),
Eq(true))
<< "expected that executor was not changed by on";
}
TEST_F(TrampolineExecutor, UsedWithVia) {
std::vector<std::string> values;
auto sender = ::folly::pushmi::make_single_sender([](auto out) {
::folly::pushmi::set_value(out, 2.0);
::folly::pushmi::set_done(out);
// ignored
::folly::pushmi::set_value(out, 1);
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
::folly::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
});
auto inlinevia = sender | op::via([&]() { return mi::inline_executor(); });
inlinevia | op::submit(v::on_value([&](auto v) {
values.push_back(folly::to<std::string>(v));
}));
EXPECT_THAT(values, ElementsAre(folly::to<std::string>(2.0)))
<< "expected that only the first item was pushed";
EXPECT_THAT(
(!std::is_same<
mi::executor_t<decltype(sender)>,
mi::executor_t<decltype(inlinevia)>>::value),
Eq(true))
<< "expected that executor was changed by via";
} }
#endif
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