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

fixes for take_until tests

Summary:
there were test failures that fell into three buckets

first, the strands worker had concurrency issues. the handoff between submit and the worker was wrong.

second, take_until was not waiting for the producer to signal done. this was originally by-design. now the expectation is that take until will not early-complete on cancellation.

third, the tests had concurrency issues. the EXPECT assertions were not allowing for non-deterministic interleaving.

Reviewed By: lewissbaker

Differential Revision: D14947473

fbshipit-source-id: 36f7cf87b9226367dc44afed663fa6418503c059
parent 1299eec8
...@@ -144,7 +144,7 @@ struct set_value_fn { ...@@ -144,7 +144,7 @@ struct set_value_fn {
std::tuple<VN...> vn_; std::tuple<VN...> vn_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue<Out, VN...>) // (requires ReceiveValue<Out, VN...>) //
void operator()(Out out) { void operator()(Out& out) {
::folly::pushmi::apply( ::folly::pushmi::apply(
::folly::pushmi::set_value, ::folly::pushmi::set_value,
std::tuple_cat(std::tuple<Out>{std::move(out)}, std::move(vn_))); std::tuple_cat(std::tuple<Out>{std::move(out)}, std::move(vn_)));
...@@ -165,7 +165,7 @@ struct set_error_fn { ...@@ -165,7 +165,7 @@ struct set_error_fn {
E e_; E e_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, E>) // (requires ReceiveError<Out, E>) //
void operator()(Out out) { void operator()(Out& out) {
set_error(out, std::move(e_)); set_error(out, std::move(e_));
} }
}; };
...@@ -183,7 +183,7 @@ struct set_done_fn { ...@@ -183,7 +183,7 @@ struct set_done_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
void operator()(Out out) { void operator()(Out& out) {
set_done(out); set_done(out);
} }
}; };
...@@ -201,7 +201,7 @@ struct set_starting_fn { ...@@ -201,7 +201,7 @@ struct set_starting_fn {
Up up_; Up up_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
void operator()(Out out) { void operator()(Out& out) {
set_starting(out, std::move(up_)); set_starting(out, std::move(up_));
} }
}; };
...@@ -308,7 +308,7 @@ struct now_fn { ...@@ -308,7 +308,7 @@ struct now_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires TimeExecutor<std::decay_t<In>>) // (requires TimeExecutor<std::decay_t<In>>) //
auto operator()(In&& in) const { auto operator()(In& in) const {
return ::folly::pushmi::now(in); return ::folly::pushmi::now(in);
} }
}; };
......
...@@ -130,8 +130,8 @@ struct flow_from_up { ...@@ -130,8 +130,8 @@ struct flow_from_up {
make_receiver([p = p_, requested](auto) { make_receiver([p = p_, requested](auto) {
auto remaining = requested; auto remaining = requested;
// this loop is structured to work when there is // this loop is structured to work when there is
// re-entrancy out.value in the loop may call up.value. // re-entrancy. out.value in the loop may call up.value.
// to handle this the state of p->c must be captured and // to handle this, the state of p->c must be captured and
// the remaining and p->c must be changed before // the remaining and p->c must be changed before
// out.value is called. // out.value is called.
while (remaining-- > 0 && !p->stop && p->c != p->end) { while (remaining-- > 0 && !p->stop && p->c != p->end) {
......
...@@ -21,35 +21,270 @@ ...@@ -21,35 +21,270 @@
#include <folly/experimental/pushmi/receiver/flow_receiver.h> #include <folly/experimental/pushmi/receiver/flow_receiver.h>
#include <folly/experimental/pushmi/receiver/receiver.h> #include <folly/experimental/pushmi/receiver/receiver.h>
#include <folly/synchronization/detail/Sleeper.h>
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
using take_until_request_channel_t = // TODO promote flow_lifetime to a general facility
//
// signals that occur in order
//
// an earlier signal should never
// happen after a later signal
//
enum class flow_progress {
invalid = 0,
construct,
start,
stop,
abort,
done,
error
};
template <class E = std::exception_ptr, class V = std::ptrdiff_t>
using flow_lifetime_request_channel_t =
any_receiver<std::exception_ptr, std::ptrdiff_t>; any_receiver<std::exception_ptr, std::ptrdiff_t>;
template <class Exec>
struct flow_lifetime_debug_on {
template<class FlowLifetime>
static void dump(FlowLifetime* that, const char* message) noexcept {
static const char* progress[] = {
"invalid", "construct", "start", "stop", "abort", "done", "error"};
printf(
"flow_lifetime: %ld values pending, %s, %s - %s\n",
that->requested_,
that->up_thread_.load() != std::thread::id{} ? "up on stack"
: "up not on stack",
that->valid() ? progress[static_cast<int>(that->progress_)] : "invalid progress",
message);
fflush(stdout);
}
};
struct flow_lifetime_debug_off {
template<class FlowLifetime>
static void dump(FlowLifetime* , const char* ) noexcept {
}
};
//
// tracks the signals that occur in during the lifetime of the
// FlowSender/FlowReceiver
//
// verify that all transitions are made safely.
//
// allow the caller to wait for the up channel
// usage to exit prior to invalidating the up channel
//
// all functions other than wait_for_up_thread() are intended to be called on
// a strand that provides the synchronization necessary
//
template <class E = std::exception_ptr, class V = std::ptrdiff_t, class Debug = flow_lifetime_debug_off>
struct flow_lifetime {
flow_progress progress_;
std::ptrdiff_t requested_;
flow_lifetime_request_channel_t<E, V> up_;
std::atomic<std::thread::id> up_thread_;
void dump(const char* message) noexcept {
Debug::template dump<flow_lifetime<E, V, Debug>>(this, message);
}
flow_lifetime() : progress_(flow_progress::construct), requested_(0) {}
bool valid() const noexcept {
return progress_ > flow_progress::invalid &&
progress_ <= flow_progress::error;
}
bool running() const noexcept {
return progress_ == flow_progress::start;
}
bool stopping() const noexcept {
return progress_ == flow_progress::stop ||
progress_ == flow_progress::abort;
}
bool stopped() const noexcept {
return progress_ >= flow_progress::done && valid();
}
bool up_valid() const noexcept {
return running() || stopping();
}
// when up_thread has not been claimed, claim it and exit.
// when up_thread has been claimed by a different thread, wait.
// when up_thread has been claimed by this thread, exit to unwind.
void wait_for_up_thread() noexcept {
const auto empty = std::thread::id{};
const auto current = std::this_thread::get_id();
auto oldValue = up_thread_.load(std::memory_order_acquire);
if (oldValue == current)
return;
folly::detail::Sleeper spin;
do {
while (oldValue != empty) {
// spin/yield
spin.wait();
oldValue = up_thread_.load(std::memory_order_acquire);
}
} while (!up_thread_.compare_exchange_weak(
oldValue,
current,
std::memory_order_release,
std::memory_order_acquire));
}
template <class Available>
std::ptrdiff_t filter_requested(Available available) const noexcept {
return running() ? std::min(std::ptrdiff_t(available), requested_) : 0;
}
void start(flow_lifetime_request_channel_t<E, V>&& up) noexcept {
if (progress_ != flow_progress::construct) {
std::terminate();
}
up_ = std::move(up);
progress_ = flow_progress::start;
}
void request(std::ptrdiff_t requested) noexcept {
if (requested < 0) {
std::terminate();
}
auto old_thread = std::thread::id{};
if (up_thread_.compare_exchange_strong(
old_thread, std::this_thread::get_id(),
std::memory_order_acquire,
std::memory_order_relaxed)) {
// done and error will be waiting while we request more from up_
if (!running()) {
std::terminate();
}
requested_ += requested;
set_value(up_, requested);
// if done and error have not exited yet
if (!stopped()) {
// signal them to exit
up_thread_.store(std::thread::id{}, std::memory_order_release);
}
}
}
void abort(E e) noexcept {
dump("abort");
auto old_thread = std::thread::id{};
if (up_thread_.compare_exchange_strong(
old_thread, std::this_thread::get_id(),
std::memory_order_acquire,
std::memory_order_relaxed)) {
// done and error will be waiting while we stop up_
if (!running()) {
std::terminate();
}
progress_ = flow_progress::abort;
set_error(up_, std::move(e));
// if done and error have not exited yet
if (!stopped()) {
// signal them to exit
up_thread_.store(std::thread::id{}, std::memory_order_release);
}
}
}
void stop() noexcept {
dump("stop");
auto old_thread = std::thread::id{};
if (up_thread_.compare_exchange_strong(
old_thread, std::this_thread::get_id(),
std::memory_order_acquire, std::memory_order_relaxed)) {
// done and error will be waiting while we stop up_
if (!running()) {
std::terminate();
}
progress_ = flow_progress::stop;
set_done(up_);
// if done and error have not exited yet
if (!stopped()) {
// signal them to exit
up_thread_.store(std::thread::id{}, std::memory_order_release);
}
}
}
void value() noexcept {
if (!running()) {
std::terminate();
}
if (--requested_ < 0) {
std::terminate();
}
}
void error() noexcept {
dump("error");
if (!running() && !stopping()) {
std::terminate();
}
progress_ = flow_progress::error;
up_ = flow_lifetime_request_channel_t<E, V>{};
}
void done() noexcept {
dump("done");
if (!running() && !stopping()) {
std::terminate();
}
progress_ = flow_progress::done;
up_ = flow_lifetime_request_channel_t<E, V>{};
}
};
template <class Debug, class Exec>
struct take_until_fn_base { struct take_until_fn_base {
Exec exec_; Exec exec_;
std::atomic<bool> done_; flow_lifetime<std::exception_ptr, std::ptrdiff_t, Debug> source_;
take_until_request_channel_t up_trigger_; flow_lifetime<std::exception_ptr, std::ptrdiff_t, Debug> trigger_;
take_until_request_channel_t up_source_; std::function<void()> cleanupFn;
explicit take_until_fn_base(Exec ex) : exec_(std::move(ex)), done_(false) {} explicit take_until_fn_base(Exec ex) : exec_(std::move(ex)) {}
}; };
template <class Exec, class Out> template <class Debug, class Exec, class Out>
struct take_until_fn_shared : public take_until_fn_base<Exec> { struct take_until_fn_shared : public take_until_fn_base<Debug, Exec> {
take_until_fn_shared(Out out, Exec exec) take_until_fn_shared(Out out, Exec exec)
: take_until_fn_base<Exec>(std::move(exec)), out_(std::move(out)) {} : take_until_fn_base<Debug, Exec>(std::move(exec)), out_(std::move(out)) {}
Out out_; Out out_;
PUSHMI_TEMPLATE(class Fn)
(requires NothrowInvocable<Fn&>&&
std::is_void<invoke_result_t<Fn&>>::value) //
void set_cleanup(Fn&& fn) {
this->source_.dump("source set_cleanup");
this->trigger_.dump("trigger set_cleanup");
if (!this->cleanupFn &&
!this->source_.stopped() && !this->trigger_.stopped()) {
this->cleanupFn = (Fn &&) fn;
}
}
void cleanup() noexcept {
this->source_.dump("source cleanup");
this->trigger_.dump("trigger cleanup");
if (!!this->cleanupFn && this->source_.stopped() &&
this->trigger_.stopped()) {
this->cleanupFn();
this->cleanupFn = nullptr;
}
}
}; };
template <class Out, class Exec> template <class Debug, class Out, class Exec>
auto make_take_until_fn_shared(Out out, Exec ex) auto make_take_until_fn_shared(Out out, Exec ex)
-> std::shared_ptr<take_until_fn_shared<Exec, Out>> { -> std::shared_ptr<take_until_fn_shared<Debug, Exec, Out>> {
return std::make_shared<take_until_fn_shared<Exec, Out>>( return std::make_shared<take_until_fn_shared<Debug, Exec, Out>>(
std::move(out), std::move(ex)); std::move(out), std::move(ex));
} }
struct take_until_debug_on : flow_lifetime_debug_on {};
struct take_until_debug_off : flow_lifetime_debug_off {};
/// The implementation of the take_until algorithms /// The implementation of the take_until algorithms
/// ///
/// This algorithm will coordinate two FlowSenders source and trigger. /// This algorithm will coordinate two FlowSenders source and trigger.
...@@ -68,243 +303,241 @@ auto make_take_until_fn_shared(Out out, Exec ex) ...@@ -68,243 +303,241 @@ auto make_take_until_fn_shared(Out out, Exec ex)
/// subject or stop_token) or any other valid expression that results in a /// subject or stop_token) or any other valid expression that results in a
/// flow_sender. /// flow_sender.
/// ///
template<class Debug>
struct take_until_fn { struct take_until_fn {
private: private:
/// The source_receiver is submitted to the source
template <class Out, class Exec>
struct source_receiver {
using shared_t = std::shared_ptr<take_until_fn_shared<Exec, Out>>;
using properties = properties_t<Out>;
using receiver_category = receiver_category_t<Out>;
shared_t s_; // TODO: promote execute() to an algorithm
template <class Fn>
template<class... VN> struct execute_receiver {
void value(VN&&... vn) { execute_receiver(Fn&& fn) : fn_((Fn &&) fn), value_(false) {}
set_value(s_->out_, (VN &&) vn...); std::decay_t<Fn> fn_;
bool value_;
using receiver_category = receiver_tag;
void value(any) {
value_ = true;
fn_();
} }
template<class E> void error(any) noexcept {
void error(E&& e) noexcept { std::terminate();
set_error(s_->out_, (E &&) e);
} }
void done() { void done() {
set_done(s_->out_); if (!value_) {
} std::terminate();
template<class Up>
void starting(Up&& up) noexcept {
set_starting(s_->out_, (Up &&) up);
} }
};
/// The trigger_receiver is submitted to the trigger
template <class Out, class Exec>
struct trigger_receiver : flow_receiver<> {
using shared_t = std::shared_ptr<take_until_fn_shared<Exec, Out>>;
shared_t s_;
explicit trigger_receiver(shared_t s) : s_(s) {}
};
/// The request_receiver is sent to the consumer when starting
template <class Out, class Exec>
struct request_receiver : receiver<> {
using shared_t = std::shared_ptr<take_until_fn_shared<Exec, Out>>;
shared_t s_;
explicit request_receiver(shared_t s) : s_(s) {}
};
/// passes source values to consumer
template <class Out>
struct on_value_impl {
template <class V>
struct impl {
V v_;
std::shared_ptr<Out> out_;
void operator()(any) {
set_value(out_, std::move(v_));
} }
}; };
template <class Data, class V>
void operator()(Data& data, V&& v) const { template <class Exec, class Fn>
if (data.s_->done_.load()) { static void execute(Exec& ex, Fn&& fn) {
return;
}
::folly::pushmi::submit( ::folly::pushmi::submit(
::folly::pushmi::schedule(data.s_->exec_), ::folly::pushmi::schedule(ex), execute_receiver<Fn>{(Fn &&) fn});
::folly::pushmi::make_receiver(impl<std::decay_t<V>>{
(V &&) v, std::shared_ptr<Out>{data.s_, &(data.s_->out_)}}));
} }
};
/// passes error to consumer and cancels trigger template <class Shared>
template <class Out> struct up_receiver_impl {
struct on_error_impl {
template<class E, class Shared>
struct impl {
E e_;
Shared s_; Shared s_;
void operator()(any) {
// cancel source
set_done(s_->up_source_);
// cancel trigger
set_done(s_->up_trigger_);
// cleanup circular references using receiver_category = receiver_tag;
s_->up_source_ = take_until_request_channel_t{};
s_->up_trigger_ = take_until_request_channel_t{};
// complete consumer /// emit request to source and trigger
set_error(s_->out_, std::move(e_)); template <class V>
void value(V&& v) const {
s_->source_.dump("source consumer request");
s_->trigger_.dump("trigger consumer request");
execute(s_->exec_, [s = s_, v = (V &&) v]() {
if (s->trigger_.running() && s->trigger_.filter_requested(1) == 0) {
s->trigger_.request(1);
} }
}; if (s->source_.running()) {
template <class Data, class E> s->source_.request(v);
void operator()(Data& data, E e) const noexcept { }
if (data.s_->done_.exchange(true)) { });
return;
} }
::folly::pushmi::submit( /// emit done to consumer after source and trigger complete
::folly::pushmi::schedule(data.s_->exec_), void error(any) noexcept {
::folly::pushmi::make_receiver( s_->source_.dump("source consumer abort");
impl<E, decltype(data.s_)>{std::move(e), data.s_})); s_->trigger_.dump("trigger consumer abort");
execute(s_->exec_, [s = s_]() {
s->set_cleanup([s]() noexcept { set_done(s->out_); });
if (s->source_.running()) {
s->source_.stop();
}
if (s->trigger_.running()) {
s->trigger_.stop();
}
});
}
/// emit done to consumer after source and trigger complete
void done() {
s_->source_.dump("source consumer stop");
s_->trigger_.dump("trigger consumer stop");
execute(s_->exec_, [s = s_]() {
s->set_cleanup([s]() noexcept { set_done(s->out_); });
if (s->source_.running()) {
s->source_.stop();
}
if (s->trigger_.running()) {
s->trigger_.stop();
}
});
} }
}; };
/// passes done to consumer and cancels trigger
template <class Out> template <class Shared>
struct on_done_impl { struct trigger_receiver_impl {
template<class Shared>
struct impl {
Shared s_; Shared s_;
void operator()(any) {
// cancel source
set_done(s_->up_source_);
// cancel trigger
set_done(s_->up_trigger_);
// cleanup circular references using receiver_category = flow_receiver_tag;
s_->up_source_ = take_until_request_channel_t{};
s_->up_trigger_ = take_until_request_channel_t{};
// complete consumer /// stop source
set_done(s_->out_); void value(any) const {
} s_->source_.dump("source trigger value");
}; s_->trigger_.dump("trigger trigger value");
template <class Data> execute(s_->exec_, [s = s_]() {
void operator()(Data& data) const { if (s->trigger_.running()) {
if (data.s_->done_.exchange(true)) { if (s->source_.running()) {
return; s->source_.stop();
} }
s->trigger_.value();
::folly::pushmi::submit( s->trigger_.stop();
::folly::pushmi::schedule(data.s_->exec_),
::folly::pushmi::make_receiver(impl<decltype(data.s_)>{data.s_}));
} }
}; });
/// passes flow requests to source
struct on_requested_impl {
struct impl {
std::ptrdiff_t requested_;
std::shared_ptr<take_until_request_channel_t> up_source_;
void operator()(any) {
// pass requested to source
set_value(up_source_, requested_);
} }
};
template <class Data, class V> /// emit done to consumer after source completes
void operator()(Data& data, V&& v) const { void error(any) noexcept {
if (data.s_->done_.load()) { s_->source_.dump("source trigger error");
return; s_->trigger_.dump("trigger trigger error");
s_->trigger_.wait_for_up_thread();
execute(s_->exec_, [s = s_]() {
s->set_cleanup([s]() noexcept { set_done(s->out_); });
if (s->source_.running()) {
s->source_.stop();
} }
::folly::pushmi::submit( s->trigger_.done();
::folly::pushmi::schedule(data.s_->exec_), s->source_.dump("source trigger error task");
::folly::pushmi::make_receiver( s->trigger_.dump("trigger trigger error task");
impl{(V &&) v, s->cleanup();
std::shared_ptr<take_until_request_channel_t>{ });
data.s_, &(data.s_->up_source_)}}));
} }
};
/// reused for all the trigger signals. cancels the source and sends done to
/// the consumer
template <class Out>
struct on_trigger_impl {
template<class Shared>
struct impl {
Shared s_;
void operator()(any) {
// cancel source
set_done(s_->up_source_);
// cancel trigger
set_done(s_->up_trigger_);
// cleanup circular references /// emit done to consumer after source completes
s_->up_source_ = take_until_request_channel_t{}; void done() {
s_->up_trigger_ = take_until_request_channel_t{}; s_->source_.dump("source trigger done");
s_->trigger_.dump("trigger trigger done");
s_->trigger_.wait_for_up_thread();
// complete consumer execute(s_->exec_, [s = s_]() {
set_done(s_->out_); s->set_cleanup([s]() noexcept { set_done(s->out_); });
if (s->source_.running()) {
s->source_.stop();
} }
}; s->trigger_.done();
s->source_.dump("source trigger done task");
s->trigger_.dump("trigger trigger done task");
template <class Data, class... AN> s->cleanup();
void operator()(Data& data, AN&&...) const noexcept { });
if (data.s_->done_.exchange(true)) {
return;
} }
// tell consumer that the end is nigh /// source and trigger are started now tell the consumer to start
::folly::pushmi::submit( template <class Up>
::folly::pushmi::schedule(data.s_->exec_), void starting(Up up) {
::folly::pushmi::make_receiver(impl<decltype(data.s_)>{data.s_})); s_->source_.dump("source trigger starting");
s_->trigger_.dump("trigger trigger starting");
execute(s_->exec_, [s = s_, up = std::move(up)]() {
s->trigger_.start(flow_lifetime_request_channel_t<>{std::move(up)});
// set back channel for consumer
set_starting(s->out_, up_receiver_impl<Shared>{s});
});
} }
}; };
/// both source and trigger are started, ask the trigger for a value and
/// give the consumer a back channel for flow control. template <class Shared, class Trigger>
template <class Out> struct source_receiver_impl {
struct on_starting_trigger_impl {
template<class Shared>
struct impl {
Shared s_; Shared s_;
void operator()(any) { Trigger t_;
// set back channel for consumer using receiver_category = flow_receiver_tag;
set_starting(
s_->out_,
::folly::pushmi::make_receiver(
request_receiver<Out, decltype(s_->exec_)>{s_},
on_requested_impl{},
on_trigger_impl<Out>{},
on_trigger_impl<Out>{}));
if (!s_->done_.load()) { /// emit value to consumer
// ask for trigger template <class V>
set_value(s_->up_trigger_, 1); void value(V&& v) {
s_->source_.dump("source source value");
s_->trigger_.dump("trigger source value");
execute(s_->exec_, [s = s_, v = (V &&) v]() mutable {
if (s->source_.running()) {
s->source_.value();
set_value(s->out_, std::move(v));
} }
});
} }
};
template <class Data, class Up>
void operator()(Data& data, Up up) const {
data.s_->up_trigger_ = take_until_request_channel_t{std::move(up)};
::folly::pushmi::submit( /// emit error to consumer after trigger completes
::folly::pushmi::schedule(data.s_->exec_), template <class E>
::folly::pushmi::make_receiver(impl<decltype(data.s_)>{data.s_})); void error(E e) noexcept {
s_->source_.dump("source source error");
s_->trigger_.dump("trigger source error");
s_->source_.wait_for_up_thread();
execute(s_->exec_, [s = s_, e = std::move(e)]() mutable {
s->set_cleanup([ s, e = std::move(e) ]() mutable noexcept {
set_error(s->out_, std::move(e));
});
if (s->trigger_.running()) {
s->trigger_.stop();
} }
}; s->source_.error();
/// source has been submitted now submit trigger s->source_.dump("source source error task");
template <class Trigger, class Out> s->trigger_.dump("trigger source error task");
struct on_starting_source_impl { s->cleanup();
Trigger t_; });
}
/// emit done to consumer after trigger completes
void done() {
s_->source_.dump("source source done");
s_->trigger_.dump("trigger source done");
s_->source_.wait_for_up_thread();
template <class Data, class Up> execute(s_->exec_, [s = s_]() {
void operator()(Data& data, Up up) { s->set_cleanup([s]() noexcept { set_done(s->out_); });
data.s_->up_source_ = take_until_request_channel_t{std::move(up)}; if (s->trigger_.running()) {
s->trigger_.stop();
}
s->source_.done();
s->source_.dump("source source done task");
s->trigger_.dump("trigger source done task");
s->cleanup();
});
}
/// source has started now submit trigger
PUSHMI_TEMPLATE(class Up)
(requires Receiver<Up>&&
FlowSenderTo<Trigger, trigger_receiver_impl<Shared>>) //
void starting(Up&& up) {
s_->source_.dump("source source starting");
s_->trigger_.dump("trigger source starting");
execute(
s_->exec_, [s = s_, t = std::move(t_), up = std::move(up)]() mutable {
s->source_.start(flow_lifetime_request_channel_t<>{std::move(up)});
// start trigger // start trigger
::folly::pushmi::submit( ::folly::pushmi::submit(
t_, std::move(t), trigger_receiver_impl<Shared>{s});
::folly::pushmi::detail::receiver_from_fn<Trigger>()( });
trigger_receiver<Out, decltype(data.s_->exec_)>{data.s_},
on_trigger_impl<Out>{},
on_trigger_impl<Out>{},
on_trigger_impl<Out>{},
on_starting_trigger_impl<Out>{}));
} }
}; };
/// submit creates a strand to use for signal coordination and submits the /// submit creates a strand to use for signal coordination and submits the
/// source /// source
template <class StrandF, class Trigger, class In> template <class StrandF, class Trigger, class In>
...@@ -313,38 +546,45 @@ struct take_until_fn { ...@@ -313,38 +546,45 @@ struct take_until_fn {
Trigger t_; Trigger t_;
PUSHMI_TEMPLATE(class SIn, class Out) PUSHMI_TEMPLATE(class SIn, class Out)
(requires Receiver<Out>) // (requires Receiver<Out>&& FlowSenderTo<
In,
source_receiver_impl<
std::shared_ptr<take_until_fn_shared<Debug, strand_t<StrandF>, Out>>,
Trigger>>) //
void void
operator()(SIn&& in, Out out) & { operator()(SIn&& in, Out out) & {
auto exec = ::folly::pushmi::make_strand(sf_); auto exec = ::folly::pushmi::make_strand(sf_);
auto s = make_take_until_fn_shared<Debug>(std::move(out), std::move(exec));
s->source_.dump("source submit");
s->trigger_.dump("trigger submit");
auto to = source_receiver_impl<decltype(s), Trigger>{std::move(s), t_};
// start source // start source
::folly::pushmi::submit( ::folly::pushmi::submit((In &&) in, std::move(to));
(In &&) in,
::folly::pushmi::detail::receiver_from_fn<In>()(
source_receiver<Out, strand_t<StrandF>>{
make_take_until_fn_shared(std::move(out), std::move(exec))},
on_value_impl<Out>{},
on_error_impl<Out>{},
on_done_impl<Out>{},
on_starting_source_impl<Trigger, Out>{t_}));
} }
PUSHMI_TEMPLATE(class SIn, class Out) PUSHMI_TEMPLATE_DEBUG(class SIn, class Out)
(requires Receiver<Out>) // (requires Receiver<Out>&& FlowSenderTo<
In,
source_receiver_impl<
std::shared_ptr<take_until_fn_shared<Debug, strand_t<StrandF>, Out>>,
Trigger>>) //
void void
operator()(SIn&& in, Out out) && { operator()(SIn&& in, Out out) && {
auto exec = ::folly::pushmi::make_strand(sf_); auto exec = ::folly::pushmi::make_strand(sf_);
auto s = make_take_until_fn_shared<Debug>(std::move(out), std::move(exec));
s->source_.dump("source submit");
s->trigger_.dump("trigger submit");
auto to = source_receiver_impl<decltype(s), Trigger>{std::move(s),
std::move(t_)};
// start source // start source
::folly::pushmi::submit( ::folly::pushmi::submit((In &&) in, std::move(to));
(In &&) in,
::folly::pushmi::detail::receiver_from_fn<In>()(
source_receiver<Out, strand_t<StrandF>>{
make_take_until_fn_shared(std::move(out), std::move(exec))},
on_value_impl<Out>{},
on_error_impl<Out>{},
on_done_impl<Out>{},
on_starting_source_impl<Trigger, Out>{std::move(t_)}));
} }
}; };
...@@ -362,12 +602,13 @@ struct take_until_fn { ...@@ -362,12 +602,13 @@ struct take_until_fn {
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
(In &&) in, submit_impl<StrandF, Trigger, In&&>{sf_, t_}); (In &&) in, submit_impl<StrandF, Trigger, In&&>{sf_, t_});
} }
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE_DEBUG(class In)
(requires FlowSender<In>) // (requires FlowSender<In>) //
auto auto
operator()(In&& in) && { operator()(In&& in) && {
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
(In &&) in, submit_impl<StrandF, Trigger, In&&>{std::move(sf_), std::move(t_)}); (In &&) in,
submit_impl<StrandF, Trigger, In&&>{std::move(sf_), std::move(t_)});
} }
}; };
...@@ -384,7 +625,8 @@ struct take_until_fn { ...@@ -384,7 +625,8 @@ struct take_until_fn {
} // namespace detail } // namespace detail
namespace operators { namespace operators {
PUSHMI_INLINE_VAR constexpr detail::take_until_fn take_until{}; PUSHMI_INLINE_VAR constexpr detail::take_until_fn<detail::take_until_debug_off> take_until{};
PUSHMI_INLINE_VAR constexpr detail::take_until_fn<detail::take_until_debug_on> debug_take_until{};
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
......
...@@ -57,7 +57,9 @@ struct ReceiverSignals_ : Base { ...@@ -57,7 +57,9 @@ struct ReceiverSignals_ : Base {
ReceiverSignals_& operator=(const ReceiverSignals_&) = default; ReceiverSignals_& operator=(const ReceiverSignals_&) = default;
ReceiverSignals_(ReceiverSignals_&&) = default; ReceiverSignals_(ReceiverSignals_&&) = default;
ReceiverSignals_& operator=(ReceiverSignals_&&) = default; ReceiverSignals_& operator=(ReceiverSignals_&&) = default;
explicit ReceiverSignals_(std::string id) : id_(std::move(id)), counters_(std::make_shared<receiver_counters>()) {} explicit ReceiverSignals_(std::string id) :
id_(std::move(id)),
counters_(std::make_shared<receiver_counters>()) {}
std::string id_; std::string id_;
std::shared_ptr<receiver_counters> counters_; std::shared_ptr<receiver_counters> counters_;
...@@ -108,6 +110,12 @@ struct ReceiverSignals_ : Base { ...@@ -108,6 +110,12 @@ struct ReceiverSignals_ : Base {
} }
} }
template<class Fn>
void verifyValues(Fn fn) {
EXPECT_THAT(fn(counters_->values_.load()), Eq(true))
<< "[" << id_
<< "]::verifyValues() expected the value signal(s) to satisfy the predicate.";
}
void verifyValues(int count) { void verifyValues(int count) {
EXPECT_THAT(counters_->values_.load(), Eq(count)) EXPECT_THAT(counters_->values_.load(), Eq(count))
<< "[" << id_ << "[" << id_
...@@ -159,6 +167,10 @@ using ReceiverSignals = ...@@ -159,6 +167,10 @@ using ReceiverSignals =
using FlowReceiverSignals = using FlowReceiverSignals =
detail::ReceiverSignals_<mi::flow_receiver<>>; detail::ReceiverSignals_<mi::flow_receiver<>>;
auto zeroOrOne = [](int count){
return count == 0 || count == 1;
};
TEST(EmptySourceEmptyTriggerTrampoline, TakeUntil) { TEST(EmptySourceEmptyTriggerTrampoline, TakeUntil) {
std::array<int, 0> ae{}; std::array<int, 0> ae{};
auto e = op::flow_from(ae, mi::trampolines); auto e = op::flow_from(ae, mi::trampolines);
...@@ -237,7 +249,7 @@ TEST(EmptySourceValueTrigger, TakeUntil) { ...@@ -237,7 +249,7 @@ TEST(EmptySourceValueTrigger, TakeUntil) {
source.verifyFinal(); source.verifyFinal();
trigger.wait(); trigger.wait();
trigger.verifyValues(1); trigger.verifyValues(zeroOrOne);
trigger.verifyDones(); trigger.verifyDones();
trigger.verifyFinal(); trigger.verifyFinal();
...@@ -263,6 +275,7 @@ TEST(ValueSourceEmptyTrigger, TakeUntil) { ...@@ -263,6 +275,7 @@ TEST(ValueSourceEmptyTrigger, TakeUntil) {
op::for_each(each); op::for_each(each);
source.wait(); source.wait();
source.verifyValues(zeroOrOne);
source.verifyDones(); source.verifyDones();
source.verifyFinal(); source.verifyFinal();
...@@ -272,7 +285,7 @@ TEST(ValueSourceEmptyTrigger, TakeUntil) { ...@@ -272,7 +285,7 @@ TEST(ValueSourceEmptyTrigger, TakeUntil) {
trigger.verifyFinal(); trigger.verifyFinal();
each.wait(); each.wait();
each.verifyValues(0); each.verifyValues(zeroOrOne);
each.verifyDones(); each.verifyDones();
each.verifyFinal(); each.verifyFinal();
} }
...@@ -291,16 +304,17 @@ TEST(ValueSourceValueTrigger, TakeUntil) { ...@@ -291,16 +304,17 @@ TEST(ValueSourceValueTrigger, TakeUntil) {
op::for_each(each); op::for_each(each);
source.wait(); source.wait();
source.verifyValues(zeroOrOne);
source.verifyDones(); source.verifyDones();
source.verifyFinal(); source.verifyFinal();
trigger.wait(); trigger.wait();
trigger.verifyValues(1); trigger.verifyValues(zeroOrOne);
trigger.verifyDones(); trigger.verifyDones();
trigger.verifyFinal(); trigger.verifyFinal();
each.wait(); each.wait();
each.verifyValues(0); each.verifyValues(zeroOrOne);
each.verifyDones(); each.verifyDones();
each.verifyFinal(); each.verifyFinal();
} }
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