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

add flow many (#49)

fbshipit-source-id: 2d6d5588ef87baf85e41ea1ccdf874a8d4f6ee66
parent d41131f9
...@@ -43,11 +43,14 @@ set(header_files ...@@ -43,11 +43,14 @@ set(header_files
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/for_each.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/from.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/from.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h"
......
...@@ -30,9 +30,9 @@ struct __pool_submit { ...@@ -30,9 +30,9 @@ struct __pool_submit {
PUSHMI_TEMPLATE(class TP, class Out) PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out>) (requires Regular<TP> && Receiver<Out>)
void operator()(TP at, Out out) const { void operator()(TP at, Out out) const {
e.execute([e = e, at = std::move(at), out = std::move(out)]() mutable { e.execute([e = make_time_single_sender(*this), at = std::move(at), out = std::move(out)]() mutable {
auto tr = trampoline(); std::this_thread::sleep_until(at);
::pushmi::submit(tr, std::move(at), std::move(out)); ::pushmi::set_value(out, e);
}); });
} }
}; };
......
This diff is collapsed.
...@@ -41,6 +41,9 @@ struct construct_deduced<many>; ...@@ -41,6 +41,9 @@ struct construct_deduced<many>;
template<> template<>
struct construct_deduced<flow_single>; struct construct_deduced<flow_single>;
template<>
struct construct_deduced<flow_many>;
template<> template<>
struct construct_deduced<sender>; struct construct_deduced<sender>;
...@@ -56,6 +59,9 @@ struct construct_deduced<flow_single_sender>; ...@@ -56,6 +59,9 @@ struct construct_deduced<flow_single_sender>;
template<> template<>
struct construct_deduced<time_single_sender>; struct construct_deduced<time_single_sender>;
template<>
struct construct_deduced<flow_many_sender>;
template <template <class...> class T, class... AN> template <template <class...> class T, class... AN>
using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>; using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>;
......
...@@ -299,10 +299,10 @@ PUSHMI_CONCEPT_DEF( ...@@ -299,10 +299,10 @@ PUSHMI_CONCEPT_DEF(
class S, class S,
class Up, class Up,
class T, class T,
class PT, class PT = std::ptrdiff_t,
class PE = std::exception_ptr, class PE = std::exception_ptr,
class E = PE) class E = PE)
(concept FlowManyReceiver)(S, Up, T, PE, E), (concept FlowManyReceiver)(S, Up, T, PT, PE, E),
ManyReceiver<S, T, E> && ManyReceiver<S, T, E> &&
ManyReceiver<Up, PT, PE> && ManyReceiver<Up, PT, PE> &&
FlowNoneReceiver<S, Up, PE, E> FlowNoneReceiver<S, Up, PE, E>
......
...@@ -242,10 +242,6 @@ struct shared_entangled : std::shared_ptr<T> { ...@@ -242,10 +242,6 @@ struct shared_entangled : std::shared_ptr<T> {
std::shared_ptr<T>(p, std::addressof(t)), dual(std::addressof(d)), lock(std::addressof(l)){ std::shared_ptr<T>(p, std::addressof(t)), dual(std::addressof(d)), lock(std::addressof(l)){
} }
shared_entangled() = delete; shared_entangled() = delete;
shared_entangled(const shared_entangled&) = delete;
shared_entangled& operator=(const shared_entangled&) = delete;
shared_entangled(shared_entangled&&) = default;
shared_entangled& operator=(shared_entangled&&) = default;
}; };
template <class First, class Second> template <class First, class Second>
...@@ -279,10 +275,6 @@ struct locked_shared_entangled_pair : std::pair<T*, Dual*> { ...@@ -279,10 +275,6 @@ struct locked_shared_entangled_pair : std::pair<T*, Dual*> {
this->second = this->e.dual; this->second = this->e.dual;
} }
locked_shared_entangled_pair() = delete; locked_shared_entangled_pair() = delete;
locked_shared_entangled_pair(const locked_shared_entangled_pair&) = delete;
locked_shared_entangled_pair& operator=(const locked_shared_entangled_pair&) = delete;
locked_shared_entangled_pair(locked_shared_entangled_pair&&) = default;
locked_shared_entangled_pair& operator=(locked_shared_entangled_pair&&) = default;
}; };
......
...@@ -22,18 +22,18 @@ void set_error(S& s, E e) noexcept(noexcept(s.error(std::move(e)))) { ...@@ -22,18 +22,18 @@ void set_error(S& s, E e) noexcept(noexcept(s.error(std::move(e)))) {
s.error(std::move(e)); s.error(std::move(e));
} }
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>().value(std::declval<V>()))) (requires requires (std::declval<S&>().value(std::declval<V&&>())))
void set_value(S& s, V&& v) noexcept(noexcept(s.value((V&&) v))) { void set_value(S& s, V&& v) noexcept(noexcept(s.value((V&&) v))) {
s.value((V&&) v); s.value((V&&) v);
} }
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>().next(std::declval<V>()))) (requires requires (std::declval<S&>().next(std::declval<V&&>())))
void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) { void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
s.next((V&&) v); s.next((V&&) v);
} }
PUSHMI_TEMPLATE (class S, class Up) PUSHMI_TEMPLATE (class S, class Up)
(requires requires (std::declval<S&>().starting(std::declval<Up>()))) (requires requires (std::declval<S&>().starting(std::declval<Up&&>())))
void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) { void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) {
s.starting((Up&&) up); s.starting((Up&&) up);
} }
...@@ -95,19 +95,19 @@ void set_error(std::reference_wrapper<S> s, E e) noexcept { ...@@ -95,19 +95,19 @@ void set_error(std::reference_wrapper<S> s, E e) noexcept {
set_error(s.get(), std::move(e)); set_error(s.get(), std::move(e));
} }
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires ( set_value(std::declval<S&>(), std::declval<V>()) )) (requires requires ( set_value(std::declval<S&>(), std::declval<V&&>()) ))
void set_value(std::reference_wrapper<S> s, V&& v) noexcept( void set_value(std::reference_wrapper<S> s, V&& v) noexcept(
noexcept(set_value(s.get(), (V&&) v))) { noexcept(set_value(s.get(), (V&&) v))) {
set_value(s.get(), (V&&) v); set_value(s.get(), (V&&) v);
} }
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires ( set_next(std::declval<S&>(), std::declval<V>()) )) (requires requires ( set_next(std::declval<S&>(), std::declval<V&&>()) ))
void set_next(std::reference_wrapper<S> s, V&& v) noexcept( void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
noexcept(set_next(s.get(), (V&&) v))) { noexcept(set_next(s.get(), (V&&) v))) {
set_next(s.get(), (V&&) v); set_next(s.get(), (V&&) v);
} }
PUSHMI_TEMPLATE (class S, class Up) PUSHMI_TEMPLATE (class S, class Up)
(requires requires ( set_starting(std::declval<S&>(), std::declval<Up>()) )) (requires requires ( set_starting(std::declval<S&>(), std::declval<Up&&>()) ))
void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept( void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept(
noexcept(set_starting(s.get(), (Up&&) up))) { noexcept(set_starting(s.get(), (Up&&) up))) {
set_starting(s.get(), (Up&&) up); set_starting(s.get(), (Up&&) up);
...@@ -163,7 +163,7 @@ struct set_error_fn { ...@@ -163,7 +163,7 @@ struct set_error_fn {
struct set_value_fn { struct set_value_fn {
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires ( (requires requires (
set_value(std::declval<S&>(), std::declval<V>()), set_value(std::declval<S&>(), std::declval<V&&>()),
set_error(std::declval<S&>(), std::current_exception()) set_error(std::declval<S&>(), std::current_exception())
)) ))
void operator()(S&& s, V&& v) const void operator()(S&& s, V&& v) const
...@@ -178,7 +178,7 @@ struct set_value_fn { ...@@ -178,7 +178,7 @@ struct set_value_fn {
struct set_next_fn { struct set_next_fn {
PUSHMI_TEMPLATE (class S, class V) PUSHMI_TEMPLATE (class S, class V)
(requires requires ( (requires requires (
set_next(std::declval<S&>(), std::declval<V>()), set_next(std::declval<S&>(), std::declval<V&&>()),
set_error(std::declval<S&>(), std::current_exception()) set_error(std::declval<S&>(), std::current_exception())
)) ))
void operator()(S&& s, V&& v) const void operator()(S&& s, V&& v) const
...@@ -194,11 +194,11 @@ struct set_next_fn { ...@@ -194,11 +194,11 @@ struct set_next_fn {
struct set_starting_fn { struct set_starting_fn {
PUSHMI_TEMPLATE (class S, class Up) PUSHMI_TEMPLATE (class S, class Up)
(requires requires ( (requires requires (
set_starting(std::declval<S&>(), std::declval<Up>()), set_starting(std::declval<S&>(), std::declval<Up&&>()),
set_error(std::declval<S&>(), std::current_exception()) set_error(std::declval<S&>(), std::current_exception())
)) ))
void operator()(S&& s, Up&& up) const void operator()(S&& s, Up&& up) const
noexcept(noexcept(set_starting(s, (Up&&) up))) { noexcept(noexcept(set_starting(s, (Up&&) up))) {
try { try {
set_starting(s, (Up&&) up); set_starting(s, (Up&&) up);
} catch (...) { } catch (...) {
......
This diff is collapsed.
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "flow_many.h"
namespace pushmi {
template <class V, class PV, class PE, class E>
class flow_many_sender<V, PV, PE, E> {
union data {
void* pobj_ = nullptr;
char buffer_[sizeof(V)]; // can hold a V in-situ
} data_{};
template <class Wrapped>
static constexpr bool insitu() {
return sizeof(Wrapped) <= sizeof(data::buffer_) &&
std::is_nothrow_move_constructible<Wrapped>::value;
}
struct vtable {
static void s_op(data&, data*) {}
static void s_submit(data&, any_flow_many<V, PV, PE, E>) {}
void (*op_)(data&, data*) = vtable::s_op;
void (*submit_)(data&, any_flow_many<V, PV, PE, E>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_;
template <class Wrapped>
flow_many_sender(Wrapped obj, std::false_type) : flow_many_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
dst->pobj_ = std::exchange(src.pobj_, nullptr);
delete static_cast<Wrapped const*>(src.pobj_);
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
static const vtable vtbl{s::op, s::submit};
data_.pobj_ = new Wrapped(std::move(obj));
vptr_ = &vtbl;
}
template <class Wrapped>
flow_many_sender(Wrapped obj, std::true_type) noexcept
: flow_many_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_),
std::move(out));
}
};
static const vtable vtbl{s::op, s::submit};
new (data_.buffer_) Wrapped(std::move(obj));
vptr_ = &vtbl;
}
template <class T, class U = std::decay_t<T>>
using wrapped_t =
std::enable_if_t<!std::is_same<U, flow_many_sender>::value, U>;
public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>;
flow_many_sender() = default;
flow_many_sender(flow_many_sender&& that) noexcept
: flow_many_sender() {
that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_);
}
PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>, is_many<>>)
explicit flow_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: flow_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~flow_many_sender() {
vptr_->op_(data_, nullptr);
}
flow_many_sender& operator=(flow_many_sender&& that) noexcept {
this->~flow_many_sender();
new ((void*)this) flow_many_sender(std::move(that));
return *this;
}
void submit(any_flow_many<V, PV, PE, E> out) {
vptr_->submit_(data_, std::move(out));
}
};
// Class static definitions:
template <class V, class PV, class PE, class E>
constexpr typename flow_many_sender<V, PV, PE, E>::vtable const
flow_many_sender<V, PV, PE, E>::noop_;
template <class SF>
class flow_many_sender<SF> {
SF sf_;
public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>;
constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(SF sf)
: sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_many<>, is_flow<>> && Invocable<SF&, Out>)
void submit(Out out) {
sf_(std::move(out));
}
};
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_many<>, is_flow<>>) Data, class DSF>
class flow_many_sender<Data, DSF> {
Data data_;
DSF sf_;
public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>;
constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(Data data)
: data_(std::move(data)) {}
constexpr flow_many_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {}
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>, is_flow<>> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
}
};
////////////////////////////////////////////////////////////////////////////////
// make_flow_many_sender
PUSHMI_INLINE_VAR constexpr struct make_flow_many_sender_fn {
inline auto operator()() const {
return flow_many_sender<ignoreSF>{};
}
PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf) const {
return flow_many_sender<SF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>, is_flow<>>)
auto operator()(Data d) const {
return flow_many_sender<Data, passDSF>{std::move(d)};
}
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>, is_flow<>>)
auto operator()(Data d, DSF sf) const {
return flow_many_sender<Data, DSF>{std::move(d), std::move(sf)};
}
} const make_flow_many_sender {};
////////////////////////////////////////////////////////////////////////////////
// deduction guides
#if __cpp_deduction_guides >= 201703
flow_many_sender() -> flow_many_sender<ignoreSF>;
PUSHMI_TEMPLATE(class SF)
(requires True<> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_many_sender(SF) -> flow_many_sender<SF>;
PUSHMI_TEMPLATE(class Data)
(requires True<> && Sender<Data, is_many<>, is_flow<>>)
flow_many_sender(Data) -> flow_many_sender<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data, is_many<>, is_flow<>>)
flow_many_sender(Data, DSF) -> flow_many_sender<Data, DSF>;
#endif
template <class V, class PV = std::ptrdiff_t, class PE = std::exception_ptr, class E = PE>
using any_flow_many_sender = flow_many_sender<V, PV, PE, E>;
template<>
struct construct_deduced<flow_many_sender>
: make_flow_many_sender_fn {};
// // TODO constrain me
// template <class V, class E = std::exception_ptr, Sender Wrapped>
// auto erase_cast(Wrapped w) {
// return flow_many_sender<V, E>{std::move(w)};
// }
} // namespace pushmi
...@@ -121,7 +121,7 @@ class flow_single_sender<Data, DSF> { ...@@ -121,7 +121,7 @@ class flow_single_sender<Data, DSF> {
DSF sf_; DSF sf_;
public: public:
using properties = property_set<is_sender<>, is_single<>>; using properties = property_set<is_sender<>, is_flow<>, is_single<>>;
constexpr flow_single_sender() = default; constexpr flow_single_sender() = default;
constexpr explicit flow_single_sender(Data data) constexpr explicit flow_single_sender(Data data)
......
...@@ -74,6 +74,12 @@ class flow_single; ...@@ -74,6 +74,12 @@ class flow_single;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN> template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_single_sender; class flow_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many_sender;
template< template<
class E = std::exception_ptr, class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point> class TP = std::chrono::system_clock::time_point>
......
...@@ -240,6 +240,7 @@ class many<Data, DNF, DEF, DDF> { ...@@ -240,6 +240,7 @@ class many<Data, DNF, DEF, DDF> {
constexpr many(Data d, DNF nf, DEF ef = DEF{}, DDF df = DDF{}) constexpr many(Data d, DNF nf, DEF ef = DEF{}, DDF df = DDF{})
: done_(false), data_(std::move(d)), nf_(nf), ef_(ef), df_(df) {} : done_(false), data_(std::move(d)), nf_(nf), ef_(ef), df_(df) {}
Data& data() {return data_;}
PUSHMI_TEMPLATE(class V) PUSHMI_TEMPLATE(class V)
(requires Invocable<DNF&, Data&, V>) (requires Invocable<DNF&, Data&, V>)
void next(V&& v) { void next(V&& v) {
......
...@@ -17,6 +17,8 @@ ...@@ -17,6 +17,8 @@
#include "../time_single_sender.h" #include "../time_single_sender.h"
#include "../flow_single.h" #include "../flow_single.h"
#include "../flow_single_sender.h" #include "../flow_single_sender.h"
#include "../flow_many.h"
#include "../flow_many_sender.h"
#include "../detail/if_constexpr.h" #include "../detail/if_constexpr.h"
#include "../detail/functional.h" #include "../detail/functional.h"
...@@ -58,6 +60,8 @@ template <> ...@@ -58,6 +60,8 @@ template <>
struct make_receiver<is_many<>> : construct_deduced<many> {}; struct make_receiver<is_many<>> : construct_deduced<many> {};
template <> template <>
struct make_receiver<is_single<>, true> : construct_deduced<flow_single> {}; struct make_receiver<is_single<>, true> : construct_deduced<flow_single> {};
template <>
struct make_receiver<is_many<>, true> : construct_deduced<flow_many> {};
template <class Cardinality, bool IsFlow> template <class Cardinality, bool IsFlow>
struct receiver_from_impl { struct receiver_from_impl {
...@@ -162,6 +166,8 @@ struct make_sender<is_many<>> : construct_deduced<many_sender> {}; ...@@ -162,6 +166,8 @@ struct make_sender<is_many<>> : construct_deduced<many_sender> {};
template <> template <>
struct make_sender<is_single<>, false, true> : construct_deduced<flow_single_sender> {}; struct make_sender<is_single<>, false, true> : construct_deduced<flow_single_sender> {};
template <> template <>
struct make_sender<is_many<>, false, true> : construct_deduced<flow_many_sender> {};
template <>
struct make_sender<is_single<>, true, false> : construct_deduced<time_single_sender> {}; struct make_sender<is_single<>, true, false> : construct_deduced<time_single_sender> {};
PUSHMI_INLINE_VAR constexpr struct sender_from_fn { PUSHMI_INLINE_VAR constexpr struct sender_from_fn {
......
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "extension_operators.h"
#include "submit.h"
namespace pushmi {
namespace detail {
struct for_each_fn {
private:
template<class... PN>
struct subset { using properties = property_set<PN...>; };
template<class In, class Out>
struct Pull : Out {
explicit Pull(Out out) : Out(std::move(out)) {}
using properties = property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>;
std::function<void(ptrdiff_t)> pull;
template<class V>
void next(V&& v) {
::pushmi::set_next(static_cast<Out&>(*this), (V&&) v);
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires ManyReceiver<Up, std::ptrdiff_t>)
void starting(Up up){
pull = [up = std::move(up)](std::ptrdiff_t requested) mutable {
::pushmi::set_next(up, requested);
};
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires None<Up> && not Many<Up>)
void starting(Up up){}
};
template <class... AN>
struct fn {
std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Flow<In> && Many<In>)
In operator()(In in) {
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_silent<>>>>()(std::move(args_))};
using Out = decltype(out);
::pushmi::submit(in, ::pushmi::detail::receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
return in;
}
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Time<In> && Flow<In> && Many<In>)
In operator()(In in) {
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_silent<>>>>()(std::move(args_))};
using Out = decltype(out);
::pushmi::submit(in, ::pushmi::now(in), ::pushmi::detail::receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
return in;
}
};
public:
template <class... AN>
auto operator()(AN&&... an) const {
return for_each_fn::fn<AN...>{{(AN&&) an...}};
}
};
} //namespace detail
namespace operators {
PUSHMI_INLINE_VAR constexpr detail::for_each_fn for_each{};
} //namespace operartors
} //namespace pushmi
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include "../many_sender.h" #include "../many_sender.h"
#include "../flow_many_sender.h"
#include "../trampoline.h"
#include "extension_operators.h" #include "extension_operators.h"
#include "submit.h" #include "submit.h"
...@@ -53,6 +55,123 @@ public: ...@@ -53,6 +55,123 @@ public:
} }
} from {}; } from {};
template <class I, class S, class Out, class Exec>
struct flow_from_producer {
flow_from_producer(I begin, S end, Out out, Exec exec, bool s) :
c(begin),
end(end),
out(std::move(out)),
exec(std::move(exec)),
stop(s) {}
I c;
S end;
Out out;
Exec exec;
std::atomic<bool> stop;
};
template<class Producer>
struct flow_from_up {
using properties = properties_t<many<>>;
explicit flow_from_up(std::shared_ptr<Producer> p) : p(std::move(p)) {}
std::shared_ptr<Producer> p;
void next(ptrdiff_t requested) {
if (requested < 1) {return;}
// submit work to exec
::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p, requested](auto) {
auto remaining = requested;
// this loop is structured to work when there is re-entrancy
// out.next in the loop may call up.next. to handle this the
// state of p->c must be captured and the remaining and p->c
// must be changed before out.next is called.
while (remaining-- > 0 && !p->stop && p->c != p->end) {
auto i = (p->c)++;
::pushmi::set_next(p->out, ::pushmi::detail::as_const(*i));
}
if (p->c == p->end) {
::pushmi::set_done(p->out);
}
}));
}
template<class E>
void error(E e) noexcept {
p->stop.store(true);
::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p](auto) {
::pushmi::set_done(p->out);
}));
}
void done() {
p->stop.store(true);
::pushmi::submit(p->exec,
::pushmi::now(p->exec),
make_single([p = p](auto) {
::pushmi::set_done(p->out);
}));
}
};
PUSHMI_INLINE_VAR constexpr struct flow_from_fn {
private:
template <class I, class S, class Exec>
struct out_impl {
I begin_;
S end_;
mutable Exec exec_;
PUSHMI_TEMPLATE(class Out)
(requires ManyReceiver<Out, typename std::iterator_traits<I>::value_type>)
void operator()(Out out) const {
using Producer = flow_from_producer<I, S, Out, Exec>;
auto p = std::make_shared<Producer>(begin_, end_, std::move(out), exec_, false);
::pushmi::submit(exec_,
::pushmi::now(exec_),
make_single([p](auto exec) {
// pass reference for cancellation.
::pushmi::set_starting(p->out, make_many(flow_from_up<Producer>{p}));
}));
}
};
public:
PUSHMI_TEMPLATE(class I, class S)
(requires
DerivedFrom<
typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag>)
auto operator()(I begin, S end) const {
return (*this)(begin, end, trampoline());
}
PUSHMI_TEMPLATE(class R)
(requires Range<R>)
auto operator()(R&& range) const {
return (*this)(std::begin(range), std::end(range), trampoline());
}
PUSHMI_TEMPLATE(class I, class S, class Exec)
(requires
DerivedFrom<
typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag> &&
Sender<Exec> && Time<Exec> && Single<Exec>)
auto operator()(I begin, S end, Exec exec) const {
return make_flow_many_sender(out_impl<I, S, Exec>{begin, end, exec});
}
PUSHMI_TEMPLATE(class R, class Exec)
(requires Range<R> && Sender<Exec> && Time<Exec> && Single<Exec>)
auto operator()(R&& range, Exec exec) const {
return (*this)(std::begin(range), std::end(range), exec);
}
} flow_from {};
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
...@@ -142,6 +142,14 @@ private: ...@@ -142,6 +142,14 @@ private:
state_->signaled.notify_all(); state_->signaled.notify_all();
} }
}; };
struct on_next_impl {
PUSHMI_TEMPLATE (class Out, class Value)
(requires Receiver<Out, is_many<>>)
void operator()(Out out, Value&& v) const {
using V = remove_cvref_t<Value>;
::pushmi::set_next(out, (Value&&) v);
}
};
struct on_error_impl { struct on_error_impl {
lock_state* state_; lock_state* state_;
PUSHMI_TEMPLATE(class Out, class E) PUSHMI_TEMPLATE(class Out, class E)
...@@ -164,6 +172,18 @@ private: ...@@ -164,6 +172,18 @@ private:
state_->signaled.notify_all(); state_->signaled.notify_all();
} }
}; };
template <bool IsTimeSender, class In>
struct submit_impl {
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
void operator()(In& in, Out out) const {
PUSHMI_IF_CONSTEXPR( (IsTimeSender) (
id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out));
) else (
id(::pushmi::submit)(in, std::move(out));
))
}
};
// TODO - only move, move-only types.. // TODO - only move, move-only types..
// if out can be copied, then submit can be called multiple // if out can be copied, then submit can be called multiple
// times.. // times..
...@@ -174,17 +194,26 @@ private: ...@@ -174,17 +194,26 @@ private:
template <bool IsTimeSender, class In> template <bool IsTimeSender, class In>
In impl_(In in) { In impl_(In in) {
lock_state state{}; lock_state state{};
auto out{::pushmi::detail::receiver_from_fn<In>()(
std::move(args_), auto submit = submit_impl<IsTimeSender, In>{};
on_value(on_value_impl{&state}), PUSHMI_IF_CONSTEXPR( ((bool)Many<In>) (
on_error(on_error_impl{&state}), auto out{::pushmi::detail::receiver_from_fn<In>()(
on_done(on_done_impl{&state}) std::move(args_),
)}; on_next(on_next_impl{}),
PUSHMI_IF_CONSTEXPR( (IsTimeSender) ( on_error(on_error_impl{&state}),
id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out)); on_done(on_done_impl{&state})
)};
submit(in, std::move(out));
) else ( ) else (
id(::pushmi::submit)(in, std::move(out)); auto out{::pushmi::detail::receiver_from_fn<In>()(
std::move(args_),
on_value(on_value_impl{&state}),
on_error(on_error_impl{&state}),
on_done(on_done_impl{&state})
)};
submit(in, std::move(out));
)) ))
std::unique_lock<std::mutex> guard{state.lock}; std::unique_lock<std::mutex> guard{state.lock};
state.signaled.wait(guard, [&]{ state.signaled.wait(guard, [&]{
return state.done; return state.done;
......
...@@ -27,15 +27,23 @@ struct tap_ { ...@@ -27,15 +27,23 @@ struct tap_ {
PUSHMI_TEMPLATE(class V, class UV = std::remove_reference_t<V>) PUSHMI_TEMPLATE(class V, class UV = std::remove_reference_t<V>)
(requires (requires
// SingleReceiver<SideEffects, const UV&> && SingleReceiver<SideEffects, const UV> &&
SingleReceiver<Out, V>) SingleReceiver<Out, UV>)
void value(V&& v) { void value(V&& v) {
::pushmi::set_value(sideEffects, as_const(v)); ::pushmi::set_value(sideEffects, as_const(v));
::pushmi::set_value(out, (V&&) v); ::pushmi::set_value(out, (V&&) v);
} }
PUSHMI_TEMPLATE(class V, class UV = std::remove_reference_t<V>)
(requires
ManyReceiver<SideEffects, const UV> &&
ManyReceiver<Out, UV>)
void next(V&& v) {
::pushmi::set_next(sideEffects, as_const(v));
::pushmi::set_next(out, (V&&) v);
}
PUSHMI_TEMPLATE(class E) PUSHMI_TEMPLATE(class E)
(requires (requires
// NoneReceiver<SideEffects, const E&> && NoneReceiver<SideEffects, const E> &&
NoneReceiver<Out, E>) NoneReceiver<Out, E>)
void error(E e) noexcept { void error(E e) noexcept {
::pushmi::set_error(sideEffects, as_const(e)); ::pushmi::set_error(sideEffects, as_const(e));
...@@ -45,6 +53,15 @@ struct tap_ { ...@@ -45,6 +53,15 @@ struct tap_ {
::pushmi::set_done(sideEffects); ::pushmi::set_done(sideEffects);
::pushmi::set_done(out); ::pushmi::set_done(out);
} }
PUSHMI_TEMPLATE(class Up, class UUp = std::remove_reference_t<Up>)
(requires
FlowReceiver<SideEffects> &&
FlowReceiver<Out>)
void starting(Up&& up) {
// up is not made const because sideEffects is allowed to call methods on up
::pushmi::set_starting(sideEffects, up);
::pushmi::set_starting(out, (Up&&) up);
}
}; };
PUSHMI_INLINE_VAR constexpr struct make_tap_fn { PUSHMI_INLINE_VAR constexpr struct make_tap_fn {
...@@ -71,12 +88,12 @@ struct tap_fn { ...@@ -71,12 +88,12 @@ struct tap_fn {
private: private:
template <class In, class SideEffects> template <class In, class SideEffects>
static auto impl(In in, SideEffects sideEffects) { static auto impl(In in, SideEffects sideEffects) {
PUSHMI_STATIC_ASSERT( // PUSHMI_STATIC_ASSERT(
::pushmi::detail::sender_requires_from<In, SideEffects, // ::pushmi::detail::sender_requires_from<In, SideEffects,
SenderTo<In, SideEffects, is_none<>>, // SenderTo<In, SideEffects, is_none<>>,
SenderTo<In, SideEffects, is_single<>>, // SenderTo<In, SideEffects, is_single<>>,
TimeSenderTo<In, SideEffects, is_single<>> >(), // TimeSenderTo<In, SideEffects, is_single<>> >(),
"'In' is not deliverable to 'SideEffects'"); // "'In' is not deliverable to 'SideEffects'");
return ::pushmi::detail::sender_from( return ::pushmi::detail::sender_from(
std::move(in), std::move(in),
...@@ -103,21 +120,21 @@ private: ...@@ -103,21 +120,21 @@ private:
PUSHMI_TEMPLATE (class Out) PUSHMI_TEMPLATE (class Out)
(requires Receiver<Out>) (requires Receiver<Out>)
auto operator()(Out out) const { auto operator()(Out out) const {
PUSHMI_STATIC_ASSERT( // PUSHMI_STATIC_ASSERT(
::pushmi::detail::sender_requires_from<In, SideEffects, // ::pushmi::detail::sender_requires_from<In, SideEffects,
SenderTo<In, Out, is_none<>>, // SenderTo<In, Out, is_none<>>,
SenderTo<In, Out, is_single<>>, // SenderTo<In, Out, is_single<>>,
TimeSenderTo<In, Out, is_single<>> >(), // TimeSenderTo<In, Out, is_single<>> >(),
"'In' is not deliverable to 'Out'"); // "'In' is not deliverable to 'Out'");
auto gang{::pushmi::detail::receiver_from_fn<In>()( auto gang{::pushmi::detail::receiver_from_fn<In>()(
detail::make_tap(sideEffects_, std::move(out)))}; detail::make_tap(sideEffects_, std::move(out)))};
using Gang = decltype(gang); using Gang = decltype(gang);
PUSHMI_STATIC_ASSERT( // PUSHMI_STATIC_ASSERT(
::pushmi::detail::sender_requires_from<In, SideEffects, // ::pushmi::detail::sender_requires_from<In, SideEffects,
SenderTo<In, Gang>, // SenderTo<In, Gang>,
SenderTo<In, Gang, is_single<>>, // SenderTo<In, Gang, is_single<>>,
TimeSenderTo<In, Gang, is_single<>> >(), // TimeSenderTo<In, Gang, is_single<>> >(),
"'In' is not deliverable to 'Out' & 'SideEffects'"); // "'In' is not deliverable to 'Out' & 'SideEffects'");
return gang; return gang;
} }
}; };
......
...@@ -15,7 +15,7 @@ namespace pushmi { ...@@ -15,7 +15,7 @@ namespace pushmi {
namespace detail { namespace detail {
template<class F, class Tag> template<class F, class Tag, bool IsFlow = false>
struct transform_on; struct transform_on;
template<class F> template<class F>
...@@ -39,6 +39,27 @@ struct transform_on<F, is_single<>> { ...@@ -39,6 +39,27 @@ struct transform_on<F, is_single<>> {
} }
}; };
template<class F>
struct transform_on<F, is_single<>, true> {
F f_;
transform_on() = default;
constexpr explicit transform_on(F f)
: f_(std::move(f)) {}
template<class Out>
auto operator()(Out out) const {
return make_flow_single(std::move(out), on_value(*this));
}
template<class Out, class V>
auto operator()(Out& out, V&& v) {
using Result = decltype(f_((V&&) v));
static_assert(::pushmi::SemiMovable<Result>,
"none of the functions supplied to transform can convert this value");
static_assert(::pushmi::Flow<Out> && ::pushmi::Single<Out>,
"Result of value transform cannot be delivered to Out");
::pushmi::set_value(out, f_((V&&) v));
}
};
template<class F> template<class F>
struct transform_on<F, is_many<>> { struct transform_on<F, is_many<>> {
F f_; F f_;
...@@ -60,6 +81,27 @@ struct transform_on<F, is_many<>> { ...@@ -60,6 +81,27 @@ struct transform_on<F, is_many<>> {
} }
}; };
template<class F>
struct transform_on<F, is_many<>, true> {
F f_;
transform_on() = default;
constexpr explicit transform_on(F f)
: f_(std::move(f)) {}
template<class Out>
auto operator()(Out out) const {
return make_flow_many(std::move(out), on_next(*this));
}
template<class Out, class V>
auto operator()(Out& out, V&& v) {
using Result = decltype(f_((V&&) v));
static_assert(::pushmi::SemiMovable<Result>,
"none of the functions supplied to transform can convert this value");
static_assert(::pushmi::Flow<Out> && ::pushmi::Many<Out>,
"Result of value transform cannot be delivered to Out");
::pushmi::set_next(out, f_((V&&) v));
}
};
struct transform_fn { struct transform_fn {
private: private:
template <class F> template <class F>
...@@ -73,7 +115,7 @@ private: ...@@ -73,7 +115,7 @@ private:
std::move(in), std::move(in),
::pushmi::detail::submit_transform_out<In>( ::pushmi::detail::submit_transform_out<In>(
// copy 'f_' to allow multiple calls to connect to multiple 'in' // copy 'f_' to allow multiple calls to connect to multiple 'in'
transform_on<F, Cardinality>{f_} transform_on<F, Cardinality, property_query_v<properties_t<In>, is_flow<>>>{f_}
) )
); );
} }
......
...@@ -131,146 +131,4 @@ namespace operators { ...@@ -131,146 +131,4 @@ namespace operators {
PUSHMI_INLINE_VAR constexpr detail::via_fn via{}; PUSHMI_INLINE_VAR constexpr detail::via_fn via{};
} // namespace operators } // namespace operators
#if 0
namespace detail {
template <class ExecutorFactory>
class fsdvia {
using executor_factory_type = std::decay_t<ExecutorFactory>;
executor_factory_type factory_;
template <class In>
class start_via {
using in_type = std::decay_t<In>;
executor_factory_type factory_;
in_type in_;
template <class Out, class Executor>
class out_via {
using out_type = std::decay_t<Out>;
using executor_type = std::decay_t<Executor>;
struct shared_type {
shared_type(out_type&& out) : out_(std::move(out)), stopped_(false) {}
out_type out_;
std::atomic_bool stopped_;
};
template <class Producer>
struct producer_proxy {
RefWrapper<Producer> up_;
std::shared_ptr<shared_type> shared_;
producer_proxy(RefWrapper<Producer> p, std::shared_ptr<shared_type> s)
: up_(std::move(p)), shared_(std::move(s)) {}
template <class V>
void value(V v) {
if (!!shared_->stopped_.exchange(true)) {
return;
}
up_.get().value(std::move(v));
}
template <class E>
void error(E e) {
if (!!shared_->stopped_.exchange(true)) {
return;
}
up_.get().error(std::move(e));
}
};
bool done_;
std::shared_ptr<shared_type> shared_;
executor_type exec_;
std::shared_ptr<AnyNone<>> upProxy_;
public:
explicit out_via(out_type&& out, executor_type&& exec)
: done_(false),
shared_(std::make_shared<shared_type>(std::move(out))),
exec_(std::move(exec)),
upProxy_() {}
template <class T>
void value(T t) {
if (done_ || shared_->stopped_) {
done_ = true;
return;
}
if (!upProxy_) {
std::abort();
}
done_ = true;
exec_ | execute([t = std::move(t), shared = shared_](auto) mutable {
shared->out_.value(std::move(t));
});
}
template <class E>
void error(E e) {
if (done_ || shared_->stopped_) {
done_ = true;
return;
}
if (!upProxy_) {
std::abort();
}
done_ = true;
exec_ | execute([e = std::move(e), shared = shared_](auto) mutable {
shared->out_.error(std::move(e));
});
}
template <class Producer>
void starting(RefWrapper<Producer> up) {
if (!!upProxy_) {
std::abort();
}
upProxy_ = std::make_shared<AnyNone<>>(AnyNone<>{
producer_proxy<Producer>{std::move(up), shared_}});
// must keep out and upProxy alive until out is notified that it is
// starting
exec_ | execute([shared = shared_, upProxy = upProxy_](auto) mutable {
shared->out_.starting(wrap_ref(*upProxy));
});
}
};
public:
start_via(executor_factory_type&& ef, in_type&& in)
: factory_(ef), in_(in) {}
template <class Out>
auto then(Out out) {
auto exec = factory_();
in_.then(out_via<Out, decltype(exec)>{std::move(out), std::move(exec)});
}
};
public:
explicit fsdvia(executor_factory_type&& ef) : factory_(std::move(ef)) {}
template <class In>
auto operator()(In in) {
return start_via<In>{std::move(factory_), std::move(in)};
}
};
} // namespace detail
namespace fsd {
template <class ExecutorFactory>
auto via(ExecutorFactory factory) {
return detail::fsdvia<ExecutorFactory>{std::move(factory)};
}
} // namespace fsd
#endif
} // namespace pushmi } // namespace pushmi
...@@ -12,6 +12,10 @@ add_executable(FlowTest FlowTest.cpp) ...@@ -12,6 +12,10 @@ add_executable(FlowTest FlowTest.cpp)
target_link_libraries(FlowTest pushmi catch Threads::Threads) target_link_libraries(FlowTest pushmi catch Threads::Threads)
catch_discover_tests(FlowTest) catch_discover_tests(FlowTest)
add_executable(FlowManyTest FlowManyTest.cpp)
target_link_libraries(FlowManyTest pushmi catch Threads::Threads)
catch_discover_tests(FlowManyTest)
add_executable(CompileTest CompileTest.cpp) add_executable(CompileTest CompileTest.cpp)
target_link_libraries(CompileTest pushmi catch Threads::Threads) target_link_libraries(CompileTest pushmi catch Threads::Threads)
# catch_discover_tests(CompileTest) # catch_discover_tests(CompileTest)
...@@ -34,7 +38,8 @@ add_executable(PushmiTest PushmiTest.cpp ...@@ -34,7 +38,8 @@ add_executable(PushmiTest PushmiTest.cpp
CompileTest.cpp CompileTest.cpp
TrampolineTest.cpp TrampolineTest.cpp
NewThreadTest.cpp NewThreadTest.cpp
FlowTest.cpp) FlowTest.cpp
FlowManyTest.cpp)
target_link_libraries(PushmiTest pushmi catch Threads::Threads) target_link_libraries(PushmiTest pushmi catch Threads::Threads)
catch_discover_tests(PushmiTest) catch_discover_tests(PushmiTest)
......
...@@ -382,3 +382,88 @@ void flow_single_sender_test(){ ...@@ -382,3 +382,88 @@ void flow_single_sender_test(){
auto any0 = pushmi::any_flow_single_sender<int>(in0); auto any0 = pushmi::any_flow_single_sender<int>(in0);
} }
void flow_many_test() {
auto out0 = pushmi::MAKE(flow_many)();
auto out1 = pushmi::MAKE(flow_many)(pushmi::ignoreVF{});
auto out2 = pushmi::MAKE(flow_many)(pushmi::ignoreVF{}, pushmi::abortEF{});
auto out3 =
pushmi::MAKE(flow_many)(
pushmi::ignoreVF{},
pushmi::abortEF{},
pushmi::ignoreDF{});
auto out4 = pushmi::MAKE(flow_many)([](auto v) { v.get(); });
auto out5 = pushmi::MAKE(flow_many)(
pushmi::on_value([](auto v) { v.get(); }, [](int v) {}),
pushmi::on_error(
[](std::exception_ptr e) noexcept{},
[](auto e) noexcept { e.get(); }
));
auto out6 = pushmi::MAKE(flow_many)(
pushmi::on_error(
[](std::exception_ptr e) noexcept {},
[](auto e) noexcept{ e.get(); }
));
auto out7 = pushmi::MAKE(flow_many)(
pushmi::on_done([]() { }));
auto out8 =
pushmi::MAKE(flow_many)(
pushmi::ignoreVF{},
pushmi::abortEF{},
pushmi::ignoreDF{},
pushmi::ignoreStrtF{});
using Out0 = decltype(out0);
auto proxy0 = pushmi::MAKE(flow_many)(out0);
auto proxy1 = pushmi::MAKE(flow_many)(out0, pushmi::passDVF{});
auto proxy2 = pushmi::MAKE(flow_many)(out0, pushmi::passDVF{}, pushmi::passDEF{});
auto proxy3 = pushmi::MAKE(flow_many)(
out0, pushmi::passDVF{}, pushmi::passDEF{}, pushmi::passDDF{});
auto proxy4 = pushmi::MAKE(flow_many)(out0, [](auto d, auto v) {
pushmi::set_value(d, v.get());
});
auto proxy5 = pushmi::MAKE(flow_many)(
out0,
pushmi::on_value([](Out0&, auto v) { v.get(); }, [](Out0&, int v) {}),
pushmi::on_error(
[](Out0&, std::exception_ptr e) noexcept {},
[](Out0&, auto e) noexcept { e.get(); }
));
auto proxy6 = pushmi::MAKE(flow_many)(
out0,
pushmi::on_error(
[](Out0&, std::exception_ptr e) noexcept {},
[](Out0&, auto e) noexcept { e.get(); }
));
auto proxy7 = pushmi::MAKE(flow_many)(
out0,
pushmi::on_done([](Out0&) { }));
auto proxy8 = pushmi::MAKE(flow_many)(out0,
pushmi::passDVF{},
pushmi::passDEF{},
pushmi::passDDF{});
auto any2 = pushmi::any_flow_many<int>(out0);
auto any3 = pushmi::any_flow_many<int>(proxy0);
}
void flow_many_sender_test(){
auto in0 = pushmi::MAKE(flow_many_sender)();
auto in1 = pushmi::MAKE(flow_many_sender)(pushmi::ignoreSF{});
auto in3 = pushmi::MAKE(flow_many_sender)([&](auto out){
in0.submit(pushmi::MAKE(flow_many)(std::move(out),
pushmi::on_value([](auto d, int v){ pushmi::set_value(d, v); })
));
});
auto out0 = pushmi::MAKE(flow_many)();
auto out1 = pushmi::MAKE(flow_many)(out0, pushmi::on_value([](auto d, int v){
pushmi::set_value(d, v);
}));
in3.submit(out1);
auto any0 = pushmi::any_flow_many_sender<int>(in0);
}
This diff is collapsed.
...@@ -204,15 +204,6 @@ SCENARIO("flow single cancellation trampoline", "[flow][sender]") { ...@@ -204,15 +204,6 @@ SCENARIO("flow single cancellation trampoline", "[flow][sender]") {
} }
} }
// copies the value of the atomic during move. requires external Synchronization
// to make the copy safe. entangle() provides the needed external
// Synchronization
template <class T>
struct moving_atomic : std::atomic<T> {
using std::atomic<T>::atomic;
moving_atomic(moving_atomic&& o) : std::atomic<T>(o.load()) {}
};
SCENARIO("flow single shared cancellation new thread", "[flow][sender]") { SCENARIO("flow single shared cancellation new thread", "[flow][sender]") {
auto nt = mi::new_thread(); auto nt = mi::new_thread();
using NT = decltype(nt); using NT = decltype(nt);
......
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