Commit 1c1fbc3c authored by Eric Niebler's avatar Eric Niebler Committed by Facebook Github Bot

add TypedSender support to pushmi, pave the way for Awaitable senders

Summary:
* The sender concepts are now defined in terms of a `sender_traits` class template, in which the user can specify additional sender metadata like what values and errors it sends to the Receiver. This will make it possible to automatically make Senders awaitable from within coroutines.

* Remove the now-unused is_sender<>, is_single<>, and is_many<> properties.

* Define a `SingleSender` concept and a `TypedSender` concept.

* Define a `pushmi::sender` class template from which a Sender can inherit to declare the sender category (single/many/flow), and the types of the values and errors.

Reviewed By: kirkshoop

Differential Revision: D14631770

fbshipit-source-id: afcfd10bcd4e0e7ba224ea2c713d7fa92249ab7f
parent 7bcdd925
...@@ -86,16 +86,19 @@ using countdownflowmany = countdown<decltype(mi::make_flow_receiver)>; ...@@ -86,16 +86,19 @@ using countdownflowmany = countdown<decltype(mi::make_flow_receiver)>;
struct inline_time_executor { struct inline_time_executor {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
std::chrono::system_clock::time_point at; : mi::single_sender_tag::with_values<inline_time_executor>::no_error {
using properties = mi::property_set< private:
mi::is_sender<>, std::chrono::system_clock::time_point at_;
mi::is_always_blocking<>, public:
mi::is_single<>>; using properties = mi::property_set<mi::is_always_blocking<>>;
task() = default;
explicit task(std::chrono::system_clock::time_point at) : at_(at) {}
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
std::this_thread::sleep_until(at); std::this_thread::sleep_until(at_);
::mi::set_value(out, inline_time_executor{}); ::mi::set_value(out, inline_time_executor{});
::mi::set_done(out); ::mi::set_done(out);
} }
...@@ -115,11 +118,8 @@ struct inline_time_executor { ...@@ -115,11 +118,8 @@ struct inline_time_executor {
struct inline_executor { struct inline_executor {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task : mi::single_sender_tag::with_values<inline_executor>::no_error {
using properties = mi::property_set< using properties = mi::property_set<mi::is_always_blocking<>>;
mi::is_sender<>,
mi::is_always_blocking<>,
mi::is_single<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
...@@ -138,22 +138,25 @@ struct inline_executor_flow_single { ...@@ -138,22 +138,25 @@ struct inline_executor_flow_single {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
CancellationFactory cf; : mi::flow_single_sender_tag::with_values<inline_executor_flow_single>
::no_error {
private:
CancellationFactory cf_;
public:
using properties = mi::property_set<mi::is_maybe_blocking<>>;
using properties = mi::property_set< task() = default;
mi::is_sender<>, explicit task(CancellationFactory cf) : cf_(std::move(cf)) {}
mi::is_flow<>,
mi::is_maybe_blocking<>,
mi::is_single<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
auto tokens = cf(); auto tokens = cf_();
using Stopper = decltype(tokens.second); using Stopper = decltype(tokens.second);
struct Data : mi::receiver<> { struct Data : mi::receiver<> {
explicit Data(Stopper stopper) : stopper(std::move(stopper)) {} explicit Data(Stopper s) : stopper(std::move(s)) {}
Stopper stopper; Stopper stopper;
}; };
auto up = mi::make_receiver( auto up = mi::make_receiver(
...@@ -173,7 +176,7 @@ struct inline_executor_flow_single { ...@@ -173,7 +176,7 @@ struct inline_executor_flow_single {
auto both = lock_both(tokens.first); auto both = lock_both(tokens.first);
if (!!both.first && !*(both.first)) { if (!!both.first && !*(both.first)) {
::mi::set_value(out, inline_executor_flow_single{cf}); ::mi::set_value(out, inline_executor_flow_single{cf_});
::mi::set_done(out); ::mi::set_done(out);
} else { } else {
// cancellation is not an error // cancellation is not an error
...@@ -191,9 +194,9 @@ struct shared_cancellation_factory { ...@@ -191,9 +194,9 @@ struct shared_cancellation_factory {
auto operator()() { auto operator()() {
// boolean cancellation // boolean cancellation
bool stop = false; bool stop = false;
auto set_stop = [](auto& stop) { auto set_stop = [](auto& stop_) {
if (!!stop) { if (!!stop_) {
*stop = true; *stop_ = true;
} }
}; };
return mi::shared_entangle(stop, set_stop); return mi::shared_entangle(stop, set_stop);
...@@ -206,9 +209,9 @@ struct entangled_cancellation_factory { ...@@ -206,9 +209,9 @@ struct entangled_cancellation_factory {
auto operator()() { auto operator()() {
// boolean cancellation // boolean cancellation
bool stop = false; bool stop = false;
auto set_stop = [](auto& stop) { auto set_stop = [](auto& stop_) {
if (!!stop) { if (!!stop_) {
*stop = true; *stop_ = true;
} }
}; };
return mi::entangle(stop, set_stop); return mi::entangle(stop, set_stop);
...@@ -220,12 +223,10 @@ using inline_executor_flow_single_entangled = ...@@ -220,12 +223,10 @@ using inline_executor_flow_single_entangled =
struct inline_executor_flow_single_ignore { struct inline_executor_flow_single_ignore {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
using properties = mi::property_set< : mi::flow_single_sender_tag::with_values<inline_executor_flow_single_ignore>
mi::is_sender<>, ::no_error {
mi::is_flow<>, using properties = mi::property_set<mi::is_maybe_blocking<>>;
mi::is_maybe_blocking<>,
mi::is_single<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
...@@ -249,27 +250,29 @@ struct inline_executor_flow_many { ...@@ -249,27 +250,29 @@ struct inline_executor_flow_many {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
: mi::flow_sender_tag::with_values<inline_executor_flow_many>
::no_error {
private:
std::atomic<int>* counter = nullptr; std::atomic<int>* counter = nullptr;
using properties = mi::property_set< public:
mi::is_sender<>, using properties = mi::property_set<mi::is_maybe_blocking<>>;
mi::is_flow<>, task() = default;
mi::is_maybe_blocking<>, explicit task(std::atomic<int>* c) : counter(c) {}
mi::is_many<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
// boolean cancellation // boolean cancellation
struct producer { struct producer {
producer(Out out, bool s) : out(std::move(out)), stop(s) {} producer(Out o, bool s) : out(std::move(o)), stop(s) {}
Out out; Out out;
std::atomic<bool> stop; std::atomic<bool> stop;
}; };
auto p = std::make_shared<producer>(std::move(out), false); auto p = std::make_shared<producer>(std::move(out), 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> sp) : p(std::move(sp)) {}
std::shared_ptr<producer> p; std::shared_ptr<producer> p;
}; };
...@@ -313,12 +316,10 @@ struct inline_executor_flow_many { ...@@ -313,12 +316,10 @@ struct inline_executor_flow_many {
struct inline_executor_flow_many_ignore { struct inline_executor_flow_many_ignore {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
using properties = mi::property_set< : mi::flow_sender_tag::with_values<inline_executor_flow_many_ignore>
mi::is_sender<>, ::no_error {
mi::is_flow<>, using properties = mi::property_set<mi::is_always_blocking<>>;
mi::is_always_blocking<>,
mi::is_many<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
// pass reference for cancellation. // pass reference for cancellation.
...@@ -336,11 +337,9 @@ struct inline_executor_flow_many_ignore { ...@@ -336,11 +337,9 @@ struct inline_executor_flow_many_ignore {
struct inline_executor_many { struct inline_executor_many {
using properties = mi::property_set<mi::is_fifo_sequence<>>; using properties = mi::property_set<mi::is_fifo_sequence<>>;
struct task { struct task
using properties = mi::property_set< : mi::sender_tag::with_values<inline_executor_many>::no_error {
mi::is_sender<>, using properties = mi::property_set<mi::is_always_blocking<>>;
mi::is_always_blocking<>,
mi::is_many<>>;
template <class Out> template <class Out>
void submit(Out out) { void submit(Out out) {
::mi::set_value(out, inline_executor_many{}); ::mi::set_value(out, inline_executor_many{});
......
...@@ -21,28 +21,51 @@ ...@@ -21,28 +21,51 @@
#include <folly/experimental/pushmi/detail/functional.h> #include <folly/experimental/pushmi/detail/functional.h>
#include <folly/experimental/pushmi/forwards.h> #include <folly/experimental/pushmi/forwards.h>
#include <folly/experimental/pushmi/properties.h> #include <folly/experimental/pushmi/properties.h>
#include <folly/experimental/pushmi/tags.h>
#include <folly/experimental/pushmi/traits.h>
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
// traits & tags // traits & tags
struct sender_tag;
struct single_sender_tag;
struct flow_sender_tag;
struct flow_single_sender_tag;
template <class PS> // add concepts to support receivers
struct has_cardinality : category_query<PS, cardinality_category> {}; //
template <class PS>
PUSHMI_INLINE_VAR constexpr bool has_cardinality_v = has_cardinality<PS>::value;
// is_single trait
template <class PS>
struct is_single<PS> : property_query<PS, is_single<>> {};
template <class PS>
PUSHMI_INLINE_VAR constexpr bool is_single_v = is_single<PS>::value;
// is_many trait /// \cond
template <class PS> namespace detail {
struct is_many<PS> : property_query<PS, is_many<>> {}; template <bool>
template <class PS> struct Enable_ {};
PUSHMI_INLINE_VAR constexpr bool is_many_v = is_many<PS>::value; template <>
struct Enable_<true> {
template <class T>
using _type = T;
};
#if defined(__cpp_fold_expressions) && __cpp_fold_expressions > 0
// An alias for the type in the pack if the pack has exactly one type in it.
// Otherwise, this SFINAE's away. T cannot be an array type of an abstract type.
template<class... Ts>
using identity_t = typename Enable_<sizeof...(Ts) == 1u>::
template _type<decltype((((Ts(*)())0)(),...))>;
#else
template <class...>
struct Front_ {};
template <class T, class... Us>
struct Front_<T, Us...> {
using type = T;
};
// Instantiation proceeds from left to right. The use of Enable_ here avoids
// needless instantiations of Front_.
template<class...Ts>
using identity_t = typename Enable_<sizeof...(Ts) == 1u>::
template _type<Front_<Ts...>>::type;
#endif
} // namespace detail
// is_flow trait // is_flow trait
template <class PS> template <class PS>
...@@ -75,32 +98,165 @@ PUSHMI_CONCEPT_DEF( ...@@ -75,32 +98,165 @@ PUSHMI_CONCEPT_DEF(
Receiver<R>&& And<MoveConstructible<VN>...>); Receiver<R>&& And<MoveConstructible<VN>...>);
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template(class R, class E = std::exception_ptr) // template(class R, class E) //
(concept ReceiveError)(R, E), // (concept ReceiveError)(R, E), //
requires(R& r, E&& e)( // requires(R& r, E&& e)( //
set_error(r, (E &&) e)) && set_error(r, (E &&) e)) &&
Receiver<R> && MoveConstructible<E> Receiver<R> && MoveConstructible<E>
); );
// is_sender trait
template <class PS>
struct is_sender<PS> : property_query<PS, is_sender<>> {};
template <class PS>
PUSHMI_INLINE_VAR constexpr bool is_sender_v = is_sender<PS>::value;
// add concepts to support senders // add concepts to support senders
// //
namespace detail {
template<template<template<class...> class, template<class...> class> class>
struct test_value_types;
template<template<template<class...> class> class>
struct test_error_type;
PUSHMI_CONCEPT_DEF(
template(class S)
concept SenderLike_,
True<typename S::sender_category>
);
PUSHMI_CONCEPT_DEF(
template(class S)
concept TypedSenderLike_,
SenderLike_<S> &&
True<test_value_types<S::template value_types>> &&
True<test_error_type<S::template error_type>>
);
template<class, class = void>
struct basic_sender_traits {
};
template<class S>
struct basic_typed_sender_traits : awaitable_senders::sender_adl_hook {
template<template<class...> class Tuple, template<class...> class Variant>
using value_types = typename S::template value_types<Tuple, Variant>;
template<template<class...> class Variant>
using error_type = typename S::template error_type<Variant>;
};
template<class S>
struct basic_sender_traits<S, std::enable_if_t<SenderLike_<S>>>
: std::conditional_t<
TypedSenderLike_<S>,
basic_typed_sender_traits<S>,
awaitable_senders::sender_adl_hook> {
using sender_category = typename S::sender_category;
};
} // namespace detail
template<typename S>
struct sender_traits
: std::conditional_t<
std::is_same<std::decay_t<S>, S>::value,
detail::basic_sender_traits<S>,
sender_traits<std::decay_t<S>>> {
using _not_specialized = void;
};
// A Sender is, by default, a "many" sender, in that it may call
// set_value on its Receiver multiple times before calling set_done.
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template(class S) // template(class S)
(concept SenderImpl)(S), // concept Sender,
SemiMovable<S>&& has_cardinality_v<S>&& SemiMovable<remove_cvref_t<S>> &&
is_sender_v<S>); True<typename sender_traits<S>::sender_category> &&
DerivedFrom<typename sender_traits<S>::sender_category, sender_tag>
);
template<class S>
PUSHMI_PP_CONSTRAINED_USING(
Sender<S>,
sender_category_t =,
typename sender_traits<S>::sender_category
);
// A single sender is a special kind of sender that promises to only
// call set_value once (or not at all) before calling set_done.
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template(class S) // template(class S)
(concept Sender)(S), // concept SingleSender,
SenderImpl<std::decay_t<S>>); Sender<S> &&
DerivedFrom<typename sender_traits<S>::sender_category, single_sender_tag>
);
PUSHMI_CONCEPT_DEF(
template(class S)
concept TypedSender,
Sender<S> &&
True<detail::test_value_types<sender_traits<S>::template value_types>> &&
True<detail::test_error_type<sender_traits<S>::template error_type>>
);
template<
class From,
template<class...> class Tuple,
template<class...> class Variant = detail::identity_t>
PUSHMI_PP_CONSTRAINED_USING(
TypedSender<From>,
sender_values_t =,
typename sender_traits<remove_cvref_t<From>>::
template value_types<Tuple, Variant>
);
template<class From, template<class...> class Variant = detail::identity_t>
PUSHMI_PP_CONSTRAINED_USING(
TypedSender<From>,
sender_error_t =,
typename sender_traits<remove_cvref_t<From>>::
template error_type<Variant>
);
namespace detail {
template<class... Ts>
using count_values = std::integral_constant<std::size_t, sizeof...(Ts)>;
} // namespace detail
PUSHMI_CONCEPT_DEF(
template(class S)
concept SingleTypedSender,
TypedSender<S> &&
SingleSender<S> &&
(sender_values_t<S, detail::count_values>::value <= 1u)
);
// /// \cond
// template<class Fun>
// struct __invoke_with {
// template<typename...Args>
// using result_t = std::invoke_result_t<Fun, Args...>;
// template<template<class...> class Tuple>
// struct as {
// template<typename...Args>
// using result_t =
// std::conditional_t<
// std::is_void_v<result_t<Args...>>,
// Tuple<>,
// Tuple<result_t<Args...>>>;
// };
// };
// /// \endcond
//
// template<class Fun, TypedSender From>
// requires requires {
// typename sender_traits<From>::template value_types<
// __invoke_with<Fun>::template result_t, __typelist>;
// }
// struct transformed_sender_of : sender_base<sender_category_t<From>> {
// template<template<class...> class Variant = identity_t>
// using error_type = sender_error_t<Variant>;
// template<
// template<class...> class Tuple,
// template<class...> class Variant = identity_t>
// using value_types =
// sender_values_t<
// From,
// __invoke_with<Fun>::template as<Tuple>::template result_t,
// Variant>;
// };
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template(class S, class R) // template(class S, class R) //
...@@ -153,7 +309,7 @@ PUSHMI_CONCEPT_DEF( ...@@ -153,7 +309,7 @@ PUSHMI_CONCEPT_DEF(
(concept Executor)(Exec), // (concept Executor)(Exec), //
requires(Exec& exec)( // requires(Exec& exec)( //
schedule(exec), schedule(exec),
requires_<Sender<decltype(schedule(exec))>>) && requires_<SingleSender<decltype(schedule(exec))>>) &&
SemiMovable<std::decay_t<Exec>>); SemiMovable<std::decay_t<Exec>>);
template <class Exec, class... Args> template <class Exec, class... Args>
...@@ -186,7 +342,8 @@ PUSHMI_CONCEPT_DEF( ...@@ -186,7 +342,8 @@ PUSHMI_CONCEPT_DEF(
template(class Exec) // template(class Exec) //
(concept StrandFactory)(Exec), // (concept StrandFactory)(Exec), //
requires(Exec& exec)( // requires(Exec& exec)( //
make_strand(exec), requires_<Strand<decltype(make_strand(exec))>>) && make_strand(exec), //
requires_<Strand<decltype(make_strand(exec))>>) &&
SemiMovable<std::decay_t<Exec>>); SemiMovable<std::decay_t<Exec>>);
template <class Exec> template <class Exec>
...@@ -210,7 +367,7 @@ PUSHMI_CONCEPT_DEF( ...@@ -210,7 +367,7 @@ PUSHMI_CONCEPT_DEF(
top(exec), top(exec),
requires_<Regular<decltype(top(exec))>>, requires_<Regular<decltype(top(exec))>>,
schedule(exec, top(exec)), schedule(exec, top(exec)),
requires_<Sender<decltype(schedule(exec, top(exec)))>>) && requires_<SingleSender<decltype(schedule(exec, top(exec)))>>) &&
Executor<Exec>); Executor<Exec>);
template <class Exec> template <class Exec>
...@@ -223,9 +380,9 @@ PUSHMI_CONCEPT_DEF( ...@@ -223,9 +380,9 @@ PUSHMI_CONCEPT_DEF(
template(class Exec, class TP, class Duration) // template(class Exec, class TP, class Duration) //
concept TimeExecutorImpl2_, // concept TimeExecutorImpl2_, //
requires(Exec& exec, TP tp, Duration d)( // requires(Exec& exec, TP tp, Duration d)( //
requires_<Sender<decltype(exec.schedule(tp + d))>>, requires_<SingleSender<decltype(exec.schedule(tp + d))>>,
requires_<Sender<decltype(exec.schedule(d + tp))>>, requires_<SingleSender<decltype(exec.schedule(d + tp))>>,
requires_<Sender<decltype(exec.schedule(tp - d))>>, requires_<SingleSender<decltype(exec.schedule(tp - d))>>,
tp += d, tp += d,
tp -= d)); tp -= d));
...@@ -247,7 +404,7 @@ PUSHMI_CONCEPT_DEF( ...@@ -247,7 +404,7 @@ PUSHMI_CONCEPT_DEF(
requires(Exec& exec)( // requires(Exec& exec)( //
now(exec), now(exec),
schedule(exec, now(exec)), schedule(exec, now(exec)),
requires_<Sender<decltype(schedule(exec, now(exec)))>>) && requires_<SingleSender<decltype(schedule(exec, now(exec)))>>) &&
ConstrainedExecutor<Exec> && ConstrainedExecutor<Exec> &&
TimeExecutorImpl_<Exec>); TimeExecutorImpl_<Exec>);
...@@ -287,7 +444,7 @@ PUSHMI_CONCEPT_DEF( ...@@ -287,7 +444,7 @@ PUSHMI_CONCEPT_DEF(
template(class S) // template(class S) //
(concept FlowSender)(S), // (concept FlowSender)(S), //
Sender<S>&& Sender<S>&&
is_flow_v<S>); DerivedFrom<sender_category_t<S>, flow_sender_tag>);
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template(class S, class R) // template(class S, class R) //
......
/*
* Copyright 2019-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <exception>
namespace folly {
namespace pushmi {
namespace awaitable_senders {
struct sender_adl_hook {
};
} // namespace awaitable_senders
// A sender inherits from
// sender_base<Category>::with_values<Values...>::with_error<Error> to
// become a TypedSender (where the Category is one of [single, many,
// flow_single, or flow_many; and where the ::with_error<Error> part is
// optional and defaults to ::with_error<std::exception_ptr>). If a sender can
// pass values As... or Bs... to a receiver, then the sender can inherit from
// sender_base<Category>::with_values<As...>::or_<Bs...>::with_error<Error>. More
// sets of alternate values can be added by tacking on more additional
// ::or_<...> instantiations; e.g.,
// sender_base<Category>::with_values<As...>::or_<Bs...>::or_<Cs...>::with_error<Error>
template <
class Category,
class BaseCategory = awaitable_senders::sender_adl_hook>
struct sender_base : BaseCategory {
using sender_category = Category;
struct no_values : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<template<class...> class, template<class...> class Variant>
using value_types = Variant<>;
template<template<class...> class Variant>
using error_type = Variant<std::exception_ptr>;
struct no_error : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<template<class...> class, template<class...> class Variant>
using value_types = Variant<>;
template<template<class...> class Variant>
using error_type = Variant<>;
};
};
template<class... Values>
struct with_values : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<template<class...> class Tuple, template<class...> class Variant>
using value_types = Variant<Tuple<Values...>>;
template<template<class...> class Variant>
using error_type = Variant<std::exception_ptr>;
struct no_error : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<template<class...> class Tuple, template<class...> class Variant>
using value_types = Variant<Tuple<Values...>>;
template<template<class...> class Variant>
using error_type = Variant<>;
};
template <class Error>
struct with_error : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<template<class...> class Tuple, template<class...> class Variant>
using value_types = value_types<Tuple, Variant>;
template<template<class...> class Variant>
using error_type = Variant<Error>;
template<typename Base, typename OtherError>
struct _or_ : private awaitable_senders::sender_adl_hook {
private:
template<template<class...> class Variant>
struct error_type_ {
template<typename... BaseErrors>
using make_variant_ = Variant<BaseErrors..., OtherError>;
};
public:
using sender_category = Category;
template<template<class...> class Tuple, template<class...> class Variant>
using value_types =
typename Base::template value_types<Tuple, Variant>;
template<template<class...> class Variant>
using error_type =
typename Base::template error_type<
error_type_<Variant>::template make_variant_>;
template<typename YetOtherError>
using or_ = _or_<_or_, YetOtherError>;
};
template<typename OtherError>
using or_ = _or_<with_error, OtherError>;
};
template<typename Base, typename...MoreValues>
struct _or_ : private awaitable_senders::sender_adl_hook {
private:
template<template<class...> class Tuple, template<class...> class Variant>
struct value_types_ {
template<typename... BaseTuples>
using make_variant_ = Variant<BaseTuples..., Tuple<MoreValues...>>;
};
public:
using sender_category = Category;
template<template<class...> class Tuple, template<class...> class Variant>
using value_types =
typename Base::template value_types<
Tuple,
value_types_<Tuple, Variant>::template make_variant_>;
template<template<class...> class Variant>
using error_type = Variant<std::exception_ptr>;
struct no_error : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<
template<class...> class Tuple,
template<class...> class Variant>
using value_types = value_types<Tuple, Variant>;
template<template<class...> class Variant>
using error_type = Variant<>;
};
template <class Error>
struct with_error : private awaitable_senders::sender_adl_hook {
using sender_category = Category;
template<
template<class...> class Tuple,
template<class...> class Variant>
using value_types = value_types<Tuple, Variant>;
template<template<class...> class Variant>
using error_type = Variant<Error>;
template<typename OtherError>
using or_ =
typename with_values::template with_error<Error>::
template _or_<with_error, OtherError>;
};
template<typename...EvenMoreValues>
using or_ = _or_<_or_, EvenMoreValues...>;
};
template<typename...MoreValues>
using or_ = _or_<with_values, MoreValues...>;
};
};
} // namespace pushmi
} // namespace folly
...@@ -48,7 +48,7 @@ int main() { ...@@ -48,7 +48,7 @@ int main() {
// matters) or the caller is just going to push the result into a queue, // matters) or the caller is just going to push the result into a queue,
// provide a way to skip the transition to a different executor and make it // provide a way to skip the transition to a different executor and make it
// stand out so that it has to be justified in code reviews. // stand out so that it has to be justified in code reviews.
mi::via_cast<mi::is_sender<>>(io_operation(io)) | op::submit(); mi::via_cast(io_operation(io)) | op::submit();
io = mi::pool_executor{}; io = mi::pool_executor{};
cpu = mi::pool_executor{}; cpu = mi::pool_executor{};
......
...@@ -67,24 +67,24 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -67,24 +67,24 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
(decltype(func)&&)func, (decltype(func)&&)func,
init(std::move(input))); init(std::move(input)));
e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable { e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable {
auto stepDone = [](auto shared_state) { auto stepDone = [](auto shared_state_) {
// pending // pending
if (--shared_state->pending_ == 0) { if (--shared_state_->pending_ == 0) {
// first exception // first exception
if (shared_state->first_exception_) { if (shared_state_->first_exception_) {
mi::set_error( mi::set_error(
shared_state->destination_, shared_state->first_exception_); shared_state_->destination_, shared_state_->first_exception_);
return; return;
} }
try { try {
// selector(accumulation) // selector(accumulation)
auto result = shared_state->selector_( auto result = shared_state_->selector_(
std::move(shared_state->accumulation_.load())); std::move(shared_state_->accumulation_.load()));
mi::set_value(shared_state->destination_, std::move(result)); mi::set_value(shared_state_->destination_, std::move(result));
mi::set_done(shared_state->destination_); mi::set_done(shared_state_->destination_);
} catch (...) { } catch (...) {
mi::set_error( mi::set_error(
shared_state->destination_, std::current_exception()); shared_state_->destination_, std::current_exception());
} }
} }
}; };
......
...@@ -35,13 +35,17 @@ struct no_fail_fn { ...@@ -35,13 +35,17 @@ struct no_fail_fn {
PUSHMI_TEMPLATE(class SIn, class Out) PUSHMI_TEMPLATE(class SIn, class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
void operator()(SIn&& in, Out out) const { void operator()(SIn&& in, Out out) const {
submit((In&&)in, ::folly::pushmi::detail::receiver_from_fn<In>()( submit(
std::move(out), ::folly::pushmi::on_error(on_error_impl{}))); (In&&)in,
receiver_from_fn<In>()(
std::move(out),
::folly::pushmi::on_error(on_error_impl{})));
} }
}; };
struct in_impl { struct in_impl {
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<In>)auto operator()(In in) const { (requires Sender<In>) //
auto operator()(In in) const {
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
std::move(in), std::move(in),
out_impl<In>{}); out_impl<In>{});
......
...@@ -39,10 +39,9 @@ public: ...@@ -39,10 +39,9 @@ public:
task schedule(); task schedule();
}; };
struct pool_executor::task { struct pool_executor::task
using properties = : single_sender_tag::with_values<pool_executor&>::no_error {
property_set< using properties = property_set<is_never_blocking<>>;
is_sender<>, is_never_blocking<>, is_single<>>;
explicit task(pool_executor e) explicit task(pool_executor e)
: pool_ex_(std::move(e)) : pool_ex_(std::move(e))
......
...@@ -69,24 +69,24 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) { ...@@ -69,24 +69,24 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
(decltype(func)&&)func, (decltype(func)&&)func,
init(std::move(input))); init(std::move(input)));
e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable { e.schedule() | op::submit([e, sb, se, shared_state](auto) mutable {
auto stepDone = [](auto shared_state) { auto stepDone = [](auto shared_state_) {
// pending // pending
if (--shared_state->pending_ == 0) { if (--shared_state_->pending_ == 0) {
// first exception // first exception
if (shared_state->first_exception_) { if (shared_state_->first_exception_) {
mi::set_error( mi::set_error(
shared_state->destination_, shared_state->first_exception_); shared_state_->destination_, shared_state_->first_exception_);
return; return;
} }
try { try {
// selector(accumulation) // selector(accumulation)
auto result = shared_state->selector_( auto result = shared_state_->selector_(
std::move(shared_state->accumulation_.load())); std::move(shared_state_->accumulation_.load()));
mi::set_value(shared_state->destination_, std::move(result)); mi::set_value(shared_state_->destination_, std::move(result));
mi::set_done(shared_state->destination_); mi::set_done(shared_state_->destination_);
} catch (...) { } catch (...) {
mi::set_error( mi::set_error(
shared_state->destination_, std::current_exception()); shared_state_->destination_, std::current_exception());
} }
} }
}; };
......
...@@ -24,7 +24,8 @@ namespace folly { ...@@ -24,7 +24,8 @@ namespace folly {
namespace pushmi { namespace pushmi {
template <class PE, class PV, class E, class... VN> template <class PE, class PV, class E, class... VN>
class any_flow_many_sender { class any_flow_many_sender
: public flow_sender_tag::with_values<VN...>::template with_error<E> {
using insitu_t = void*[2]; using insitu_t = void*[2];
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
...@@ -83,7 +84,7 @@ class any_flow_many_sender { ...@@ -83,7 +84,7 @@ class any_flow_many_sender {
using wrapped_t = using wrapped_t =
std::enable_if_t<!std::is_same<U, any_flow_many_sender>::value, U>; std::enable_if_t<!std::is_same<U, any_flow_many_sender>::value, U>;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>; using properties = property_set<>;
any_flow_many_sender() = default; any_flow_many_sender() = default;
any_flow_many_sender(any_flow_many_sender&& that) noexcept any_flow_many_sender(any_flow_many_sender&& that) noexcept
...@@ -92,7 +93,7 @@ class any_flow_many_sender { ...@@ -92,7 +93,7 @@ class any_flow_many_sender {
std::swap(that.vptr_, vptr_); std::swap(that.vptr_, vptr_);
} }
PUSHMI_TEMPLATE (class Wrapped) PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>> && is_many_v<wrapped_t<Wrapped>>) (requires FlowSender<wrapped_t<Wrapped>>)
explicit any_flow_many_sender(Wrapped obj) noexcept(insitu<Wrapped>()) explicit any_flow_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: any_flow_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {} : any_flow_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_flow_many_sender() { ~any_flow_many_sender() {
...@@ -120,7 +121,8 @@ class flow_many_sender<SF> { ...@@ -120,7 +121,8 @@ class flow_many_sender<SF> {
SF sf_; SF sf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>; using sender_category = flow_sender_tag;
using properties = property_set<>;
constexpr flow_many_sender() = default; constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(SF sf) constexpr explicit flow_many_sender(SF sf)
...@@ -133,20 +135,18 @@ class flow_many_sender<SF> { ...@@ -133,20 +135,18 @@ class flow_many_sender<SF> {
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender) Data, class DSF> template <PUSHMI_TYPE_CONSTRAINT(FlowSender) Data, class DSF>
class flow_many_sender<Data, DSF> { class flow_many_sender<Data, DSF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
public: public:
using sender_category = flow_sender_tag;
using properties = properties_t<Data>; using properties = properties_t<Data>;
static_assert( static_assert(
FlowSender<Data>, FlowSender<Data>,
"Data must be a flow sender"); "Data must be a flow sender");
static_assert(
is_many_v<Data>,
"Data must be a many sender");
constexpr flow_many_sender() = default; constexpr flow_many_sender() = default;
constexpr explicit flow_many_sender(Data data) constexpr explicit flow_many_sender(Data data)
...@@ -187,12 +187,12 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_many_sender_fn { ...@@ -187,12 +187,12 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_many_sender_fn {
return flow_many_sender<SF>{std::move(sf)}; return flow_many_sender<SF>{std::move(sf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && FlowSender<Data> && is_many_v<Data>) (requires True<> && FlowSender<Data>)
auto operator()(Data d) const { auto operator()(Data d) const {
return flow_many_sender<Data, passDSF>{std::move(d)}; return flow_many_sender<Data, passDSF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires FlowSender<Data> && is_many_v<Data>) (requires FlowSender<Data>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return flow_many_sender<Data, DSF>{std::move(d), std::move(sf)}; return flow_many_sender<Data, DSF>{std::move(d), std::move(sf)};
} }
...@@ -208,11 +208,11 @@ PUSHMI_TEMPLATE(class SF) ...@@ -208,11 +208,11 @@ PUSHMI_TEMPLATE(class SF)
flow_many_sender(SF) -> flow_many_sender<SF>; flow_many_sender(SF) -> flow_many_sender<SF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && FlowSender<Data> && is_many_v<Data>) (requires True<> && FlowSender<Data>)
flow_many_sender(Data) -> flow_many_sender<Data, passDSF>; flow_many_sender(Data) -> flow_many_sender<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires FlowSender<Data> && is_many_v<Data>) (requires FlowSender<Data>)
flow_many_sender(Data, DSF) -> flow_many_sender<Data, DSF>; flow_many_sender(Data, DSF) -> flow_many_sender<Data, DSF>;
#endif #endif
......
...@@ -25,7 +25,8 @@ namespace folly { ...@@ -25,7 +25,8 @@ namespace folly {
namespace pushmi { namespace pushmi {
template <class PE, class E, class... VN> template <class PE, class E, class... VN>
class any_flow_single_sender { class any_flow_single_sender
: public flow_single_sender_tag::with_values<VN...>::template with_error<E> {
using insitu_t = void*[2]; using insitu_t = void*[2];
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
...@@ -90,7 +91,7 @@ class any_flow_single_sender { ...@@ -90,7 +91,7 @@ class any_flow_single_sender {
using wrapped_t = using wrapped_t =
std::enable_if_t<!std::is_same<U, any_flow_single_sender>::value, U>; std::enable_if_t<!std::is_same<U, any_flow_single_sender>::value, U>;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_single<>>; using properties = property_set<>;
any_flow_single_sender() = default; any_flow_single_sender() = default;
any_flow_single_sender(any_flow_single_sender&& that) noexcept any_flow_single_sender(any_flow_single_sender&& that) noexcept
...@@ -99,7 +100,8 @@ class any_flow_single_sender { ...@@ -99,7 +100,8 @@ class any_flow_single_sender {
std::swap(that.vptr_, vptr_); std::swap(that.vptr_, vptr_);
} }
PUSHMI_TEMPLATE (class Wrapped) PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>> &&is_single_v<wrapped_t<Wrapped>>) (requires SingleSender<wrapped_t<Wrapped>> &&
FlowSender<wrapped_t<Wrapped>>)
explicit any_flow_single_sender(Wrapped obj) noexcept(insitu<Wrapped>()) explicit any_flow_single_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: any_flow_single_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {} : any_flow_single_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_flow_single_sender() { ~any_flow_single_sender() {
...@@ -113,7 +115,9 @@ class any_flow_single_sender { ...@@ -113,7 +115,9 @@ class any_flow_single_sender {
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, E>&& ReceiveValue<Out, VN...>) // (requires ReceiveError<Out, E>&& ReceiveValue<Out, VN...>) //
void submit(Out&& out) { void submit(Out&& out) {
vptr_->submit_(data_, any_flow_receiver<PE, std::ptrdiff_t, E, VN...>{(Out &&) out}); vptr_->submit_(
data_,
any_flow_receiver<PE, std::ptrdiff_t, E, VN...>{(Out &&) out});
} }
}; };
...@@ -127,7 +131,8 @@ class flow_single_sender<SF> { ...@@ -127,7 +131,8 @@ class flow_single_sender<SF> {
SF sf_; SF sf_;
public: public:
using properties = property_set<is_sender<>, is_flow<>, is_single<>>; using sender_category = flow_single_sender_tag;
using properties = property_set<>;
constexpr flow_single_sender() = default; constexpr flow_single_sender() = default;
constexpr explicit flow_single_sender(SF sf) constexpr explicit flow_single_sender(SF sf)
...@@ -141,20 +146,21 @@ class flow_single_sender<SF> { ...@@ -141,20 +146,21 @@ class flow_single_sender<SF> {
}; };
template < template <
PUSHMI_TYPE_CONSTRAINT(Sender) Data, PUSHMI_TYPE_CONSTRAINT(SingleSender) Data,
class DSF> class DSF>
class flow_single_sender<Data, DSF> { class flow_single_sender<Data, DSF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
public: public:
using sender_category = flow_single_sender_tag;
using properties = properties_t<Data>; using properties = properties_t<Data>;
static_assert( static_assert(
FlowSender<Data>, FlowSender<Data>,
"Data must be a flow sender"); "Data must be a flow sender");
static_assert( static_assert(
is_single_v<Data>, SingleSender<Data>,
"Data must be a single sender"); "Data must be a single sender");
constexpr flow_single_sender() = default; constexpr flow_single_sender() = default;
...@@ -190,12 +196,12 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn { ...@@ -190,12 +196,12 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn {
return flow_single_sender<SF>{std::move(sf)}; return flow_single_sender<SF>{std::move(sf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && FlowSender<Data> && is_single_v<Data>) (requires True<> && FlowSender<Data> && SingleSender<Data>)
auto operator()(Data d) const { auto operator()(Data d) const {
return flow_single_sender<Data, passDSF>{std::move(d)}; return flow_single_sender<Data, passDSF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires FlowSender<Data> && is_single_v<Data>) (requires FlowSender<Data> && SingleSender<Data>)
auto operator()(Data d, DSF sf) const { auto operator()(Data d, DSF sf) const {
return flow_single_sender<Data, DSF>{std::move(d), std::move(sf)}; return flow_single_sender<Data, DSF>{std::move(d), std::move(sf)};
} }
...@@ -211,11 +217,11 @@ PUSHMI_TEMPLATE(class SF) ...@@ -211,11 +217,11 @@ PUSHMI_TEMPLATE(class SF)
flow_single_sender(SF) -> flow_single_sender<SF>; flow_single_sender(SF) -> flow_single_sender<SF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<> && FlowSender<Data> && is_single_v<Data>) (requires True<> && FlowSender<Data> && SingleSender<Data>)
flow_single_sender(Data) -> flow_single_sender<Data, passDSF>; flow_single_sender(Data) -> flow_single_sender<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires FlowSender<Data> && is_single_v<Data>) (requires FlowSender<Data> && SingleSender<Data>)
flow_single_sender(Data, DSF) -> flow_single_sender<Data, DSF>; flow_single_sender(Data, DSF) -> flow_single_sender<Data, DSF>;
#endif #endif
......
...@@ -25,12 +25,14 @@ class inline_constrained_executor_t { ...@@ -25,12 +25,14 @@ class inline_constrained_executor_t {
using properties = property_set<is_fifo_sequence<>>; using properties = property_set<is_fifo_sequence<>>;
template<class CV> template<class CV>
struct task { struct task
: single_sender_tag::with_values<inline_constrained_executor_t>::no_error {
private:
CV cv_; CV cv_;
using properties = property_set< public:
is_sender<>, task() = default;
is_always_blocking<>, explicit task(CV cv) : cv_(cv) {}
is_single<>>; using properties = property_set<is_always_blocking<>>;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
...@@ -39,15 +41,15 @@ class inline_constrained_executor_t { ...@@ -39,15 +41,15 @@ class inline_constrained_executor_t {
set_done(out); set_done(out);
} }
}; };
auto top() { std::ptrdiff_t top() {
return 0; return 0;
} }
task<std::ptrdiff_t> schedule() { auto schedule() {
return {top()}; return task<std::ptrdiff_t>{top()};
} }
template<class CV> template <class CV>
task<std::ptrdiff_t> schedule(CV cv) { auto schedule(CV cv) {
return {cv}; return task<CV>{cv};
} }
}; };
...@@ -66,12 +68,13 @@ class inline_time_executor_t { ...@@ -66,12 +68,13 @@ class inline_time_executor_t {
using properties = property_set<is_fifo_sequence<>>; using properties = property_set<is_fifo_sequence<>>;
template<class TP> template<class TP>
struct task { struct task : single_sender_tag::with_values<inline_time_executor_t>::no_error {
private:
TP tp_; TP tp_;
using properties = property_set< public:
is_sender<>, using properties = property_set<is_always_blocking<>>;
is_always_blocking<>, task() = default;
is_single<>>; explicit task(TP tp) : tp_(std::move(tp)) {}
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
...@@ -85,12 +88,12 @@ class inline_time_executor_t { ...@@ -85,12 +88,12 @@ class inline_time_executor_t {
auto top() { auto top() {
return std::chrono::system_clock::now(); return std::chrono::system_clock::now();
} }
task<std::chrono::system_clock::time_point> schedule() { auto schedule() {
return {top()}; return task<std::chrono::system_clock::time_point>{top()};
} }
template<class TP> template<class TP>
task<TP> schedule(TP tp) { auto schedule(TP tp) {
return {tp}; return task<TP>{tp};
} }
}; };
...@@ -108,11 +111,8 @@ class inline_executor_t { ...@@ -108,11 +111,8 @@ class inline_executor_t {
public: public:
using properties = property_set<is_fifo_sequence<>>; using properties = property_set<is_fifo_sequence<>>;
struct task { struct task : single_sender_tag::with_values<inline_executor_t>::no_error {
using properties = property_set< using properties = property_set<is_always_blocking<>>;
is_sender<>,
is_always_blocking<>,
is_single<>>;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
...@@ -128,7 +128,7 @@ class inline_executor_t { ...@@ -128,7 +128,7 @@ class inline_executor_t {
}; };
struct inlineEXF { struct inlineEXF {
inline_executor_t operator()() { inline_executor_t operator()() const {
return {}; return {};
} }
}; };
......
...@@ -24,7 +24,8 @@ namespace folly { ...@@ -24,7 +24,8 @@ namespace folly {
namespace pushmi { namespace pushmi {
template <class E, class... VN> template <class E, class... VN>
class any_many_sender { class any_many_sender
: public sender_tag::with_values<VN...>::template with_error<E> {
using insitu_t = void*[2]; using insitu_t = void*[2];
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
...@@ -83,7 +84,7 @@ class any_many_sender { ...@@ -83,7 +84,7 @@ class any_many_sender {
std::enable_if_t<!std::is_same<U, any_many_sender>::value, U>; std::enable_if_t<!std::is_same<U, any_many_sender>::value, U>;
public: public:
using properties = property_set<is_sender<>, is_many<>>; using properties = property_set<>;
any_many_sender() = default; any_many_sender() = default;
any_many_sender(any_many_sender&& that) noexcept : any_many_sender() { any_many_sender(any_many_sender&& that) noexcept : any_many_sender() {
...@@ -92,10 +93,8 @@ class any_many_sender { ...@@ -92,10 +93,8 @@ class any_many_sender {
} }
PUSHMI_TEMPLATE(class Wrapped) PUSHMI_TEMPLATE(class Wrapped)
(requires SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>> && (requires SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>>) //
is_many_v<wrapped_t<Wrapped>>) // explicit any_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
explicit any_many_sender(Wrapped obj) //
noexcept(insitu<Wrapped>())
: any_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {} : any_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_many_sender() { ~any_many_sender() {
vptr_->op_(data_, nullptr); vptr_->op_(data_, nullptr);
...@@ -118,11 +117,11 @@ constexpr typename any_many_sender<E, VN...>::vtable const ...@@ -118,11 +117,11 @@ constexpr typename any_many_sender<E, VN...>::vtable const
any_many_sender<E, VN...>::noop_; any_many_sender<E, VN...>::noop_;
template <class SF> template <class SF>
class many_sender<SF> { class many_sender<SF> : public sender_tag {
SF sf_; SF sf_;
public: public:
using properties = property_set<is_sender<>, is_many<>>; using properties = property_set<>;
constexpr many_sender() = default; constexpr many_sender() = default;
constexpr explicit many_sender(SF sf) : sf_(std::move(sf)) {} constexpr explicit many_sender(SF sf) : sf_(std::move(sf)) {}
...@@ -141,14 +140,12 @@ class many_sender<Data, DSF> { ...@@ -141,14 +140,12 @@ class many_sender<Data, DSF> {
DSF sf_; DSF sf_;
public: public:
using sender_category = sender_tag;
using properties = properties_t<Data>; using properties = properties_t<Data>;
static_assert( static_assert(
Sender<Data>, Sender<Data>,
"Data must be a sender"); "Data must be a sender");
static_assert(
is_many_v<Data>,
"Data must be a many sender");
constexpr many_sender() = default; constexpr many_sender() = default;
constexpr explicit many_sender(Data data) : data_(std::move(data)) {} constexpr explicit many_sender(Data data) : data_(std::move(data)) {}
...@@ -188,13 +185,13 @@ PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn { ...@@ -188,13 +185,13 @@ PUSHMI_INLINE_VAR constexpr struct make_many_sender_fn {
return many_sender<SF>{std::move(sf)}; return many_sender<SF>{std::move(sf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<>&& Sender<Data> && is_many_v<Data>) // (requires True<>&& Sender<Data>) //
auto auto
operator()(Data d) const { operator()(Data d) const {
return many_sender<Data, passDSF>{std::move(d)}; return many_sender<Data, passDSF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data> && is_many_v<Data>) // (requires Sender<Data>) //
auto auto
operator()(Data d, DSF sf) const { operator()(Data d, DSF sf) const {
return many_sender<Data, DSF>{std::move(d), std::move(sf)}; return many_sender<Data, DSF>{std::move(d), std::move(sf)};
...@@ -212,12 +209,12 @@ PUSHMI_TEMPLATE(class SF) ...@@ -212,12 +209,12 @@ PUSHMI_TEMPLATE(class SF)
->many_sender<SF>; ->many_sender<SF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<>&& Sender<Data> && is_many_v<Data>) // (requires True<>&& Sender<Data>) //
many_sender(Data) many_sender(Data)
->many_sender<Data, passDSF>; ->many_sender<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data> && is_many_v<Data>) // (requires Sender<Data>) //
many_sender(Data, DSF) many_sender(Data, DSF)
->many_sender<Data, DSF>; ->many_sender<Data, DSF>;
#endif #endif
......
...@@ -26,11 +26,9 @@ namespace pushmi { ...@@ -26,11 +26,9 @@ namespace pushmi {
struct new_thread_executor; struct new_thread_executor;
struct new_thread_task { struct new_thread_task
using properties = property_set< : single_sender_tag::with_values<any_executor_ref<std::exception_ptr>> {
is_sender<>, using properties = property_set<is_never_blocking<>>;
is_never_blocking<>,
is_single<>>;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) (requires Receiver<Out>)
......
...@@ -28,28 +28,31 @@ namespace operators { ...@@ -28,28 +28,31 @@ namespace operators {
PUSHMI_INLINE_VAR constexpr struct defer_fn { PUSHMI_INLINE_VAR constexpr struct defer_fn {
private: private:
template <class F> template <class F>
struct impl { struct task : sender_traits<invoke_result_t<F&>> {
using sender_type = invoke_result_t<F&>;
F f_; F f_;
PUSHMI_TEMPLATE(class Data, class Out) public:
(requires Receiver<Out>) using properties = properties_t<sender_type>;
void operator()(Data&, Out out) {
task() = default;
explicit task(F f) : f_(std::move(f)) {}
PUSHMI_TEMPLATE(class Out)
(requires SenderTo<sender_type&, Out>)
void submit(Out out) {
auto sender = f_(); auto sender = f_();
submit(sender, std::move(out)); pushmi::submit(sender, std::move(out));
} }
}; };
public: public:
PUSHMI_TEMPLATE(class F) PUSHMI_TEMPLATE(class F)
(requires Invocable<F&>) (requires Invocable<F&> && Sender<invoke_result_t<F&>>)
auto operator()(F f) const { auto operator()(F f) const {
struct sender_base : single_sender<> { return task<F>{std::move(f)};
using properties = properties_t<invoke_result_t<F&>>;
};
return make_single_sender(sender_base{}, impl<F>{std::move(f)});
} }
} defer{}; } defer{};
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
} // namespace folly } // namespace folly
...@@ -21,25 +21,23 @@ ...@@ -21,25 +21,23 @@
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
namespace detail { namespace operators {
struct single_empty_impl : pipeorigin { PUSHMI_INLINE_VAR constexpr struct empty_fn {
using properties = property_set< private:
is_sender<>, struct task : pipeorigin, single_sender_tag::with_values<> {
is_single<>, using properties = property_set<is_always_blocking<>>;
is_always_blocking<>>;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>&& Invocable<decltype(set_done)&, Out&>) // (requires Receiver<Out>) //
void submit(Out&& out) { void submit(Out&& out) {
set_done(out); set_done(out);
} }
}; };
} // namespace detail public:
auto operator()() const {
namespace operators { return task{};
inline detail::single_empty_impl empty() { }
return {}; } empty {};
}
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
......
...@@ -20,35 +20,33 @@ ...@@ -20,35 +20,33 @@
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
namespace detail { namespace operators {
struct single_error_sender_base : single_sender<ignoreSF> {
using properties = property_set< PUSHMI_INLINE_VAR constexpr struct error_fn {
is_sender<>, private:
is_single<>, template <class E>
is_always_blocking<>>; struct task : single_sender_tag::with_values<>::with_error<E> {
}; private:
template <class E>
struct single_error_impl {
E e_; E e_;
PUSHMI_TEMPLATE(class Base, class Out) public:
using properties = property_set<is_always_blocking<>>;
task() = default;
explicit task(E e) : e_(std::move(e)) {}
PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, E>) (requires ReceiveError<Out, E>)
void operator()( void submit(Out&& out) && {
Base&&,
Out&& out) {
set_error(out, std::move(e_)); set_error(out, std::move(e_));
} }
}; };
} // namespace detail
namespace operators { public:
PUSHMI_TEMPLATE(class E)
PUSHMI_TEMPLATE(class E) (requires SemiMovable<E>)
(requires MoveConstructible<E>) auto operator()(E e) const {
auto error(E e) { return task<E>{std::move(e)};
return make_single_sender( }
detail::single_error_sender_base{}, } error {};
detail::single_error_impl<E>{std::move(e)});
}
} // namespace operators } // namespace operators
} // namespace pushmi } // namespace pushmi
......
...@@ -82,65 +82,53 @@ struct receiver_from_impl { ...@@ -82,65 +82,53 @@ struct receiver_from_impl {
using MakeReceiver = make_receiver<IsFlow>; using MakeReceiver = make_receiver<IsFlow>;
template <class... AN> template <class... AN>
using receiver_type = ::folly::pushmi::invoke_result_t<MakeReceiver&, AN...>; using receiver_type = ::folly::pushmi::invoke_result_t<MakeReceiver&, AN...>;
PUSHMI_TEMPLATE(class... Ts) PUSHMI_TEMPLATE(class... Ts)
(requires Invocable<MakeReceiver&, Ts...>) // (requires Invocable<MakeReceiver&, Ts...>) //
auto auto operator()(std::tuple<Ts...> args) const {
operator()(std::tuple<Ts...> args) const {
return ::folly::pushmi::apply(MakeReceiver(), std::move(args)); return ::folly::pushmi::apply(MakeReceiver(), std::move(args));
} }
PUSHMI_TEMPLATE(
class... Ts, PUSHMI_TEMPLATE(class... Ts, class F0, class... Fns)
class F0, (requires And<SemiMovable<F0>, SemiMovable<Fns>...>&&
class... Fns) Invocable<MakeReceiver&, Ts...>&&
(requires And<SemiMovable<F0>, SemiMovable<Fns>...>&& Invocable<MakeReceiver&, Ts...>&& Invocable<const receiver_from_impl&, receiver_type<Ts...>, F0, Fns...>) //
Invocable< auto operator()(std::tuple<Ts...> args, F0 f0, Fns... fns) const {
const receiver_from_impl&,
receiver_type<Ts...>,
F0, Fns...>) //
auto
operator()(std::tuple<Ts...> args, F0 f0, Fns... fns) const {
return (*this)((*this)(std::move(args)), std::move(f0), std::move(fns)...); return (*this)((*this)(std::move(args)), std::move(f0), std::move(fns)...);
} }
PUSHMI_TEMPLATE(class Out, class... Fns) PUSHMI_TEMPLATE(class Out, class... Fns)
(requires not is_v<Out, std::tuple>&& And<MoveConstructible<Fns&&>...>) // (requires not is_v<Out, std::tuple>&& And<MoveConstructible<Fns&&>...>) //
auto auto operator()(Out&& out, Fns&&... fns) const {
operator()(Out&& out, Fns&&... fns) const {
return MakeReceiver()((Out&&)out, (Fns&&)fns...); return MakeReceiver()((Out&&)out, (Fns&&)fns...);
} }
}; };
template <PUSHMI_TYPE_CONSTRAINT(Sender) In> template <PUSHMI_TYPE_CONSTRAINT(Sender) In>
using receiver_from_fn = receiver_from_impl< using receiver_from_fn = receiver_from_impl<FlowSender<In>>;
property_query_v<properties_t<In>, is_flow<>>>;
template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class... AN> template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class... AN>
using receiver_type_t = using receiver_type_t =
typename receiver_from_fn<In>::template receiver_type<AN...>; typename receiver_from_fn<In>::template receiver_type<AN...>;
template < template <class SenderCategory>
class Cardinality,
bool IsFlow = false>
struct make_sender; struct make_sender;
template <> template <>
struct make_sender<is_single<>> : construct_deduced<single_sender> {}; struct make_sender<single_sender_tag> : construct_deduced<single_sender> {};
template <> template <>
struct make_sender<is_many<>> : construct_deduced<many_sender> {}; struct make_sender<sender_tag> : construct_deduced<many_sender> {};
template <> template <>
struct make_sender<is_single<>, true> struct make_sender<flow_single_sender_tag>
: construct_deduced<flow_single_sender> {}; : construct_deduced<flow_single_sender> {};
template <> template <>
struct make_sender<is_many<>, true> struct make_sender<flow_sender_tag>
: construct_deduced<flow_many_sender> {}; : construct_deduced<flow_many_sender> {};
PUSHMI_INLINE_VAR constexpr struct sender_from_fn { PUSHMI_INLINE_VAR constexpr struct sender_from_fn {
PUSHMI_TEMPLATE(class In, class... FN) PUSHMI_TEMPLATE(class In, class... FN)
(requires Sender<In>) // (requires Sender<In>) //
auto auto operator()(In in, FN&&... fn) const {
operator()(In in, FN&&... fn) const { return make_sender<sender_category_t<In>>{}(std::move(in), (FN &&) fn...);
using MakeSender = make_sender<
property_set_index_t<properties_t<In>, is_single<>>,
property_query_v<properties_t<In>, is_flow<>>>;
return MakeSender{}(std::move(in), (FN &&) fn...);
} }
} const sender_from{}; } const sender_from{};
...@@ -151,8 +139,7 @@ struct set_value_fn { ...@@ -151,8 +139,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 void operator()(Out out) {
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_)));
...@@ -173,8 +160,7 @@ struct set_error_fn { ...@@ -173,8 +160,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 void operator()(Out out) {
operator()(Out out) {
set_error(out, std::move(e_)); set_error(out, std::move(e_));
} }
}; };
...@@ -182,8 +168,7 @@ struct set_error_fn { ...@@ -182,8 +168,7 @@ struct set_error_fn {
public: public:
PUSHMI_TEMPLATE(class E) PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>) // (requires SemiMovable<E>) //
auto auto operator()(E e) const {
operator()(E e) const {
return impl<E>{std::move(e)}; return impl<E>{std::move(e)};
} }
}; };
...@@ -193,8 +178,7 @@ struct set_done_fn { ...@@ -193,8 +178,7 @@ struct set_done_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
void void operator()(Out out) {
operator()(Out out) {
set_done(out); set_done(out);
} }
}; };
...@@ -212,8 +196,7 @@ struct set_starting_fn { ...@@ -212,8 +196,7 @@ struct set_starting_fn {
Up up_; Up up_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
void void operator()(Out out) {
operator()(Out out) {
set_starting(out, std::move(up_)); set_starting(out, std::move(up_));
} }
}; };
...@@ -221,8 +204,7 @@ struct set_starting_fn { ...@@ -221,8 +204,7 @@ struct set_starting_fn {
public: public:
PUSHMI_TEMPLATE(class Up) PUSHMI_TEMPLATE(class Up)
(requires Receiver<Up>) // (requires Receiver<Up>) //
auto auto operator()(Up up) const {
operator()(Up up) const {
return impl<Up>{std::move(up)}; return impl<Up>{std::move(up)};
} }
}; };
...@@ -232,8 +214,7 @@ struct get_executor_fn { ...@@ -232,8 +214,7 @@ struct get_executor_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<In>) // (requires Sender<In>) //
auto auto operator()(In& in) const {
operator()(In& in) const {
return get_executor(in); return get_executor(in);
} }
}; };
...@@ -249,8 +230,7 @@ struct make_strand_fn { ...@@ -249,8 +230,7 @@ struct make_strand_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<In>) // (requires Sender<In>) //
auto auto operator()(In& in) const {
operator()(In& in) const {
return make_strand(in); return make_strand(in);
} }
}; };
...@@ -268,14 +248,12 @@ struct do_submit_fn { ...@@ -268,14 +248,12 @@ struct do_submit_fn {
Out out_; Out out_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires SenderTo<In, Out>) // (requires SenderTo<In, Out>) //
void void operator()(In&& in) && {
operator()(In&& in) && {
submit((In&&)in, std::move(out_)); submit((In&&)in, std::move(out_));
} }
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires SenderTo<In, Out>) // (requires SenderTo<In, Out>) //
void void operator()(In&& in) & {
operator()(In&& in) & {
submit((In&&)in, out_); submit((In&&)in, out_);
} }
}; };
...@@ -283,8 +261,7 @@ struct do_submit_fn { ...@@ -283,8 +261,7 @@ struct do_submit_fn {
public: public:
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>) // (requires Receiver<Out>) //
auto auto operator()(Out out) const {
operator()(Out out) const {
return impl<Out>{std::move(out)}; return impl<Out>{std::move(out)};
} }
}; };
...@@ -294,8 +271,7 @@ struct do_schedule_fn { ...@@ -294,8 +271,7 @@ struct do_schedule_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class Ex, class... VN) PUSHMI_TEMPLATE(class Ex, class... VN)
(requires Executor<Ex>) // (requires Executor<Ex>) //
auto auto operator()(Ex& ex, VN&&... vn) {
operator()(Ex& ex, VN&&... vn) {
return schedule(ex, (VN&&)vn...); return schedule(ex, (VN&&)vn...);
} }
}; };
...@@ -311,8 +287,7 @@ struct top_fn { ...@@ -311,8 +287,7 @@ struct top_fn {
struct impl { struct impl {
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires ConstrainedExecutor<std::decay_t<In>>) // (requires ConstrainedExecutor<std::decay_t<In>>) //
auto auto operator()(In& in) const {
operator()(In& in) const {
return ::folly::pushmi::top(in); return ::folly::pushmi::top(in);
} }
}; };
...@@ -328,8 +303,7 @@ struct now_fn { ...@@ -328,8 +303,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 auto operator()(In&& in) const {
operator()(In&& in) const {
return ::folly::pushmi::now(in); return ::folly::pushmi::now(in);
} }
}; };
......
...@@ -25,10 +25,6 @@ namespace detail { ...@@ -25,10 +25,6 @@ namespace detail {
struct for_each_fn { struct for_each_fn {
private: private:
template <class... PN>
struct subset {
using properties = property_set<PN...>;
};
template<class Up> template<class Up>
struct request_fn { struct request_fn {
Up up_; Up up_;
...@@ -75,17 +71,16 @@ struct for_each_fn { ...@@ -75,17 +71,16 @@ struct for_each_fn {
struct fn { struct fn {
std::tuple<AN...> args_; std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires FlowSender<In>&& is_many_v<In>) (requires FlowSender<In>)
In operator()(In in) { In operator()(In in) {
auto out{::folly::pushmi::detail::receiver_from_fn<subset< // Strip flow:
is_sender<>, using C =
property_set_index_t<properties_t<In>, is_single<>>>>()( std::conditional_t<SingleSender<In>, single_sender_tag, sender_tag>;
std::move(args_))}; auto out{receiver_from_fn<C>()(std::move(args_))};
using Out = decltype(out); using Out = decltype(out);
::folly::pushmi::submit( ::folly::pushmi::submit(
in, in,
::folly::pushmi::detail::receiver_from_fn<In>()( receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
Pull<In, Out>{std::move(out)}));
return in; return in;
} }
}; };
......
...@@ -28,29 +28,32 @@ PUSHMI_CONCEPT_DEF( ...@@ -28,29 +28,32 @@ PUSHMI_CONCEPT_DEF(
template(class R) // template(class R) //
concept Range, // concept Range, //
requires(R&& r)( // requires(R&& r)( //
implicitly_convertible_to<bool>(std::begin(r) == std::end(r)))); implicitly_convertible_to<bool>(std::begin(r) == std::end(r))
)
);
namespace operators { namespace operators {
PUSHMI_INLINE_VAR constexpr struct from_fn { PUSHMI_INLINE_VAR constexpr struct from_fn {
private: private:
template <class I, class S> template <class I, class S>
struct sender_impl : pipeorigin { struct task
using properties = property_set< : sender_tag::with_values<typename std::iterator_traits<I>::value_type>
is_sender<>, ::no_error {
is_many<>, private:
is_always_blocking<>>;
I begin_; I begin_;
S end_; S end_;
sender_impl(I begin, S end) : begin_(begin), end_(end) {} public:
using properties = property_set<is_always_blocking<>>;
task() = default;
task(I begin, S end) : begin_(begin), end_(end) {}
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue< (requires ReceiveValue<
Out, Out,
typename std::iterator_traits<I>::value_type>) // typename std::iterator_traits<I>::value_type>) //
void void submit(Out&& out) const {
submit(Out out) { for (auto c = begin_; c != end_; ++c) {
auto c = begin_;
for (; c != end_; ++c) {
set_value(out, *c); set_value(out, *c);
} }
set_done(out); set_done(out);
...@@ -62,15 +65,13 @@ PUSHMI_INLINE_VAR constexpr struct from_fn { ...@@ -62,15 +65,13 @@ PUSHMI_INLINE_VAR constexpr struct from_fn {
(requires DerivedFrom< (requires DerivedFrom<
typename std::iterator_traits<I>::iterator_category, typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag>) // std::forward_iterator_tag>) //
auto auto operator()(I begin, S end) const {
operator()(I begin, S end) const { return task<I, S>{begin, end};
return sender_impl<I, S>{begin, end};
} }
PUSHMI_TEMPLATE(class R) PUSHMI_TEMPLATE(class R)
(requires Range<R>) // (requires Range<R>) //
auto auto operator()(R&& range) const {
operator()(R&& range) const {
return (*this)(std::begin(range), std::end(range)); return (*this)(std::begin(range), std::end(range));
} }
} from{}; } from{};
...@@ -177,25 +178,28 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn { ...@@ -177,25 +178,28 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn {
void done() { void done() {
} }
}; };
template <class I, class S, class Exec> template <class I, class S, class Exec>
struct sender_impl : pipeorigin { struct task
using properties = property_set< : flow_sender_tag::with_values<typename std::iterator_traits<I>::value_type>
is_sender<>, ::no_error
is_flow<>, , pipeorigin {
is_many<>, using properties = property_set<is_always_blocking<>>;
is_always_blocking<>>;
I begin_; I begin_;
S end_; S end_;
Exec exec_; Exec exec_;
sender_impl(I begin, S end, Exec exec) : begin_(begin), end_(end), exec_(exec) {}
task(I begin, S end, Exec exec)
: begin_(begin), end_(end), exec_(exec) {
}
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue< (requires ReceiveValue<
Out&, Out&,
typename std::iterator_traits<I>::value_type>) // typename std::iterator_traits<I>::value_type>) //
void void submit(Out out) {
submit(Out out) {
using Producer = flow_from_producer<I, S, Out, Exec>; using Producer = flow_from_producer<I, S, Out, Exec>;
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);
...@@ -210,15 +214,13 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn { ...@@ -210,15 +214,13 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn {
(requires DerivedFrom< (requires DerivedFrom<
typename std::iterator_traits<I>::iterator_category, typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag>) // std::forward_iterator_tag>) //
auto auto operator()(I begin, S end) const {
operator()(I begin, S end) const {
return (*this)(begin, end, trampoline()); return (*this)(begin, end, trampoline());
} }
PUSHMI_TEMPLATE(class R) PUSHMI_TEMPLATE(class R)
(requires Range<R>) // (requires Range<R>) //
auto auto operator()(R&& range) const {
operator()(R&& range) const {
return (*this)(std::begin(range), std::end(range), trampoline()); return (*this)(std::begin(range), std::end(range), trampoline());
} }
...@@ -226,15 +228,13 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn { ...@@ -226,15 +228,13 @@ PUSHMI_INLINE_VAR constexpr struct flow_from_fn {
(requires DerivedFrom< (requires DerivedFrom<
typename std::iterator_traits<I>::iterator_category, typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag>&& Executor<Exec>) // std::forward_iterator_tag>&& Executor<Exec>) //
auto auto operator()(I begin, S end, Exec exec) const {
operator()(I begin, S end, Exec exec) const { return task<I, S, Exec>{begin, end, exec};
return sender_impl<I, S, Exec>{begin, end, exec};
} }
PUSHMI_TEMPLATE(class R, class Exec) PUSHMI_TEMPLATE(class R, class Exec)
(requires Range<R>&& Executor<Exec>) // (requires Range<R>&& Executor<Exec>) //
auto auto operator()(R&& range, Exec exec) const {
operator()(R&& range, Exec exec) const {
return (*this)(std::begin(range), std::end(range), exec); return (*this)(std::begin(range), std::end(range), exec);
} }
} flow_from{}; } flow_from{};
......
...@@ -17,7 +17,6 @@ ...@@ -17,7 +17,6 @@
#include <folly/experimental/pushmi/o/extension_operators.h> #include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h> #include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/single_sender.h>
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
...@@ -26,22 +25,33 @@ namespace operators { ...@@ -26,22 +25,33 @@ namespace operators {
PUSHMI_INLINE_VAR constexpr struct just_fn { PUSHMI_INLINE_VAR constexpr struct just_fn {
private: private:
struct sender_base : single_sender<> {
using properties = property_set<
is_sender<>,
is_single<>,
is_always_blocking<>>;
};
template <class... VN> template <class... VN>
struct impl { struct task : single_sender_tag::with_values<VN...>::no_error {
private:
std::tuple<VN...> vn_; std::tuple<VN...> vn_;
PUSHMI_TEMPLATE(class In, class Out) public:
(requires ReceiveValue<Out, VN...>) // using properties = property_set<is_always_blocking<>>;
void
operator()(In&&, Out&& out) { task() = default;
explicit task(VN&&... vn) : vn_{(VN&&) vn...} {}
PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue<Out&, VN...>) //
void submit(Out&& out) && {
::folly::pushmi::apply(
[&out](VN&&... vn) { set_value(out, (VN&&) vn...); },
std::move(vn_)
);
set_done(out);
}
PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue<Out&, VN&...>) //
void submit(Out&& out) & {
::folly::pushmi::apply( ::folly::pushmi::apply(
::folly::pushmi::set_value, [&out](VN&... vn) { set_value(out, vn...); },
std::tuple_cat(std::tuple<Out&>{out}, std::move(vn_))); vn_
);
set_done(out); set_done(out);
} }
}; };
...@@ -49,10 +59,8 @@ PUSHMI_INLINE_VAR constexpr struct just_fn { ...@@ -49,10 +59,8 @@ PUSHMI_INLINE_VAR constexpr struct just_fn {
public: public:
PUSHMI_TEMPLATE(class... VN) PUSHMI_TEMPLATE(class... VN)
(requires And<SemiMovable<VN>...>) // (requires And<SemiMovable<VN>...>) //
auto auto operator()(VN... vn) const {
operator()(VN... vn) const { return task<VN...>{std::move(vn)...};
return make_single_sender(
sender_base{}, impl<VN...>{std::tuple<VN...>{std::move(vn)...}});
} }
} just{}; } just{};
} // namespace operators } // namespace operators
......
...@@ -52,7 +52,7 @@ struct on_fn { ...@@ -52,7 +52,7 @@ struct on_fn {
struct adapt_impl { struct adapt_impl {
Factory ef_; Factory ef_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<std::decay_t<In>>) // (requires Sender<In>) //
auto auto
operator()(In&& in) const { operator()(In&& in) const {
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
......
...@@ -62,14 +62,13 @@ PUSHMI_INLINE_VAR constexpr detail::request_via_fn request_via{}; ...@@ -62,14 +62,13 @@ PUSHMI_INLINE_VAR constexpr detail::request_via_fn request_via{};
} // namespace operators } // namespace operators
PUSHMI_TEMPLATE(class To, class In) PUSHMI_TEMPLATE(class In)
(requires Same<To, is_sender<>>&& Sender<In>) (requires Sender<In>)
auto via_cast(In in) { auto via_cast(In in) {
return in; return in;
} }
PUSHMI_TEMPLATE(class To, class In) template<class In>
(requires Same<To, is_sender<>>)
auto via_cast(send_via<In> ss) { auto via_cast(send_via<In> ss) {
return ss.in; return ss.in;
} }
......
...@@ -127,20 +127,22 @@ struct blocking_submit_fn { ...@@ -127,20 +127,22 @@ struct blocking_submit_fn {
} }
auto schedule() { auto schedule() {
auto protected_scope = protect_stack{state_}; auto protected_scope = protect_stack{state_};
return nested_task_impl<decltype(::folly::pushmi::schedule(ex_))>{ return nested_task_impl<sender_t<Exec>>{
state_, ::folly::pushmi::schedule(ex_)}; state_,
::folly::pushmi::schedule(ex_)};
} }
PUSHMI_TEMPLATE(class Exec_ = Exec) PUSHMI_TEMPLATE(class Exec_ = Exec)
(requires ConstrainedExecutor<Exec_>) // (requires ConstrainedExecutor<Exec_>) //
auto schedule(constraint_t<Exec_> at) { auto schedule(constraint_t<Exec_> at) {
auto protected_scope = protect_stack{state_}; auto protected_scope = protect_stack{state_};
return nested_task_impl<sender_t<Exec_, constraint_t<Exec_>>>{ return nested_task_impl<sender_t<Exec_, constraint_t<Exec_>>>{
state_, ::folly::pushmi::schedule(ex_, std::move(at))}; state_,
::folly::pushmi::schedule(ex_, std::move(at))};
} }
}; };
template<class Task> template<class Task>
struct nested_task_impl { struct nested_task_impl : sender_traits<Task> {
nested_task_impl(lock_state* state, Task t) nested_task_impl(lock_state* state, Task t)
: state_(state), t_(std::move(t)) {} : state_(state), t_(std::move(t)) {}
lock_state* state_; lock_state* state_;
...@@ -154,7 +156,8 @@ struct blocking_submit_fn { ...@@ -154,7 +156,8 @@ struct blocking_submit_fn {
auto protected_scope = protect_stack{state_}; auto protected_scope = protect_stack{state_};
++state_->nested; // reversed in nested_receiver_impl ::done/::error ++state_->nested; // reversed in nested_receiver_impl ::done/::error
::folly::pushmi::submit( ::folly::pushmi::submit(
std::move(t_), nested_receiver_impl<Out>{state_, std::move(out)}); std::move(t_),
nested_receiver_impl<Out>{state_, std::move(out)});
} }
}; };
...@@ -271,7 +274,7 @@ struct blocking_submit_fn { ...@@ -271,7 +274,7 @@ struct blocking_submit_fn {
std::tuple<AN...> args_; std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<std::decay_t<In>>&& Invocable< (requires Sender<In>&& Invocable<
submit_impl<In>&, submit_impl<In>&,
In&&, In&&,
::folly::pushmi::invoke_result_t< ::folly::pushmi::invoke_result_t<
......
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
PUSHMI_TEMPLATE(class SideEffects, class Out) PUSHMI_TEMPLATE(class SideEffects, class Out)
...@@ -74,9 +73,7 @@ struct tap_fn { ...@@ -74,9 +73,7 @@ struct tap_fn {
struct impl_fn { struct impl_fn {
PUSHMI_TEMPLATE(class In, class SideEffects) PUSHMI_TEMPLATE(class In, class SideEffects)
(requires SenderTo<In, SideEffects>) // (requires SenderTo<In, SideEffects>) //
auto operator()( auto operator()(In&& in, SideEffects&& sideEffects) {
In&& in,
SideEffects&& sideEffects) {
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
(In &&) in, (In &&) in,
submit_impl<In, std::decay_t<SideEffects>>{(SideEffects &&) submit_impl<In, std::decay_t<SideEffects>>{(SideEffects &&)
...@@ -88,17 +85,16 @@ struct tap_fn { ...@@ -88,17 +85,16 @@ struct tap_fn {
struct adapt_impl { struct adapt_impl {
std::tuple<AN...> args_; std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<std::decay_t<In>>) // (requires Sender<In>) //
auto operator()(In&& in) { auto operator()(In&& in) {
return tap_fn::impl_fn{}( return tap_fn::impl_fn{}(
(In &&) in, (In &&) in,
::folly::pushmi::detail::receiver_from_fn<In>()(std::move(args_))); receiver_from_fn<In>()(std::move(args_)));
} }
}; };
PUSHMI_TEMPLATE(class In, class SideEffects) PUSHMI_TEMPLATE(class In, class SideEffects)
(requires Sender<std::decay_t<In>>&& (requires Sender<In> && Receiver<SideEffects>) //
Receiver<SideEffects>) //
struct submit_impl { struct submit_impl {
SideEffects sideEffects_; SideEffects sideEffects_;
template <class Out> template <class Out>
...@@ -113,7 +109,7 @@ struct tap_fn { ...@@ -113,7 +109,7 @@ struct tap_fn {
&& SenderTo<In, Out>&& && SenderTo<In, Out>&&
SenderTo<In, receiver_t<Out>>) // SenderTo<In, receiver_t<Out>>) //
auto operator()(Data&& in, Out&& out) const { auto operator()(Data&& in, Out&& out) const {
auto gang{::folly::pushmi::detail::receiver_from_fn<std::decay_t<In>>()( auto gang{receiver_from_fn<std::decay_t<In>>()(
detail::make_tap(sideEffects_, (Out &&) out))}; detail::make_tap(sideEffects_, (Out &&) out))};
submit((In &&) in, std::move(gang)); submit((In &&) in, std::move(gang));
} }
......
...@@ -25,11 +25,11 @@ namespace pushmi { ...@@ -25,11 +25,11 @@ namespace pushmi {
namespace detail { namespace detail {
template <class F, class Tag, bool IsFlow = false> template <class F, class SenderCategory>
struct transform_on; struct transform_on;
template <class F> template <class F>
struct transform_on<F, is_single<>> { struct transform_on<F, single_sender_tag> {
F f_; F f_;
transform_on() = default; transform_on() = default;
constexpr explicit transform_on(F f) : f_(std::move(f)) {} constexpr explicit transform_on(F f) : f_(std::move(f)) {}
...@@ -56,7 +56,7 @@ struct transform_on<F, is_single<>> { ...@@ -56,7 +56,7 @@ struct transform_on<F, is_single<>> {
}; };
template <class F> template <class F>
struct transform_on<F, is_single<>, true> { struct transform_on<F, flow_single_sender_tag> {
F f_; F f_;
transform_on() = default; transform_on() = default;
constexpr explicit transform_on(F f) : f_(std::move(f)) {} constexpr explicit transform_on(F f) : f_(std::move(f)) {}
...@@ -78,7 +78,7 @@ struct transform_on<F, is_single<>, true> { ...@@ -78,7 +78,7 @@ struct transform_on<F, is_single<>, true> {
}; };
template <class F> template <class F>
struct transform_on<F, is_many<>> { struct transform_on<F, sender_tag> {
F f_; F f_;
transform_on() = default; transform_on() = default;
constexpr explicit transform_on(F f) : f_(std::move(f)) {} constexpr explicit transform_on(F f) : f_(std::move(f)) {}
...@@ -100,7 +100,7 @@ struct transform_on<F, is_many<>> { ...@@ -100,7 +100,7 @@ struct transform_on<F, is_many<>> {
}; };
template <class F> template <class F>
struct transform_on<F, is_many<>, true> { struct transform_on<F, flow_sender_tag> {
F f_; F f_;
transform_on() = default; transform_on() = default;
constexpr explicit transform_on(F f) : f_(std::move(f)) {} constexpr explicit transform_on(F f) : f_(std::move(f)) {}
...@@ -128,16 +128,11 @@ struct transform_fn { ...@@ -128,16 +128,11 @@ struct transform_fn {
F f_; F f_;
PUSHMI_TEMPLATE(class SIn, class Out) PUSHMI_TEMPLATE(class SIn, class Out)
(requires Receiver<std::decay_t<Out>>) // (requires Receiver<std::decay_t<Out>>) //
auto auto operator()(SIn&& in, Out&& out) const {
operator()(SIn&& in, Out&& out) const {
using Cardinality = property_set_index_t<properties_t<In>, is_single<>>;
// copy 'f_' to allow multiple calls to connect to multiple 'in' // copy 'f_' to allow multiple calls to connect to multiple 'in'
::folly::pushmi::submit( ::folly::pushmi::submit(
(In &&) in, (In &&) in,
transform_on< transform_on<F, sender_category_t<In>>{f_}((Out &&) out));
F,
Cardinality,
property_query_v<properties_t<In>, is_flow<>>>{f_}((Out &&) out));
} }
}; };
...@@ -145,9 +140,8 @@ struct transform_fn { ...@@ -145,9 +140,8 @@ struct transform_fn {
struct adapt_impl { struct adapt_impl {
F f_; F f_;
PUSHMI_TEMPLATE(class In) PUSHMI_TEMPLATE(class In)
(requires Sender<std::decay_t<In>>) // (requires Sender<In>) //
auto auto operator()(In&& in) const {
operator()(In&& in) const {
// copy 'f_' to allow multiple calls to connect to multiple 'in' // copy 'f_' to allow multiple calls to connect to multiple 'in'
return ::folly::pushmi::detail::sender_from( return ::folly::pushmi::detail::sender_from(
(In &&) in, submit_impl<F, In&&>{f_}); (In &&) in, submit_impl<F, In&&>{f_});
......
...@@ -515,7 +515,7 @@ template<> ...@@ -515,7 +515,7 @@ template<>
struct construct_deduced<receiver> : make_receiver_fn {}; struct construct_deduced<receiver> : make_receiver_fn {};
PUSHMI_TEMPLATE (class T, class In) PUSHMI_TEMPLATE (class T, class In)
(requires SenderTo<In, std::promise<T>> && is_single_v<In>) (requires SenderTo<In, std::promise<T>> && SingleSender<In>)
std::future<T> future_from(In&& in) { std::future<T> future_from(In&& in) {
std::promise<T> p; std::promise<T> p;
auto result = p.get_future(); auto result = p.get_future();
...@@ -523,7 +523,7 @@ std::future<T> future_from(In&& in) { ...@@ -523,7 +523,7 @@ std::future<T> future_from(In&& in) {
return result; return result;
} }
PUSHMI_TEMPLATE (class In) PUSHMI_TEMPLATE (class In)
(requires SenderTo<In, std::promise<void>> && is_single_v<In>) (requires SenderTo<In, std::promise<void>> && SingleSender<In>)
std::future<void> future_from(In&& in) { std::future<void> future_from(In&& in) {
std::promise<void> p; std::promise<void> p;
auto result = p.get_future(); auto result = p.get_future();
......
...@@ -24,7 +24,8 @@ namespace folly { ...@@ -24,7 +24,8 @@ namespace folly {
namespace pushmi { namespace pushmi {
template <class E, class... VN> template <class E, class... VN>
class any_single_sender { class any_single_sender
: public single_sender_tag::with_values<VN...>::template with_error<E> {
using insitu_t = void*[2]; using insitu_t = void*[2];
union data { union data {
void* pobj_ = nullptr; void* pobj_ = nullptr;
...@@ -44,7 +45,8 @@ class any_single_sender { ...@@ -44,7 +45,8 @@ class any_single_sender {
static constexpr vtable const noop_{}; static constexpr vtable const noop_{};
vtable const* vptr_ = &noop_; vtable const* vptr_ = &noop_;
template <class Wrapped> template <class Wrapped>
any_single_sender(Wrapped obj, std::true_type, std::false_type) : any_single_sender() { any_single_sender(Wrapped obj, std::true_type, std::false_type)
: any_single_sender() {
struct s { struct s {
static void op(data& src, data* dst) { static void op(data& src, data* dst) {
if (dst) if (dst)
...@@ -119,7 +121,7 @@ class any_single_sender { ...@@ -119,7 +121,7 @@ class any_single_sender {
using wrapped_t = using wrapped_t =
std::enable_if_t<!std::is_same<U, any_single_sender>::value, U>; std::enable_if_t<!std::is_same<U, any_single_sender>::value, U>;
public: public:
using properties = property_set<is_sender<>, is_single<>>; using properties = property_set<>;
any_single_sender() = default; any_single_sender() = default;
any_single_sender(any_single_sender&& that) noexcept : any_single_sender() { any_single_sender(any_single_sender&& that) noexcept : any_single_sender() {
...@@ -128,7 +130,8 @@ class any_single_sender { ...@@ -128,7 +130,8 @@ class any_single_sender {
} }
PUSHMI_TEMPLATE(class Wrapped) // PUSHMI_TEMPLATE(class Wrapped) //
(requires SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>>) // (requires SingleSender<wrapped_t<Wrapped>> &&
SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>>) //
explicit any_single_sender(Wrapped&& obj) noexcept(insitu<Wrapped>()) explicit any_single_sender(Wrapped&& obj) noexcept(insitu<Wrapped>())
: any_single_sender{std::move(obj), std::is_rvalue_reference<Wrapped&&>{}, bool_<insitu<Wrapped>()>{}} {} : any_single_sender{std::move(obj), std::is_rvalue_reference<Wrapped&&>{}, bool_<insitu<Wrapped>()>{}} {}
~any_single_sender() { ~any_single_sender() {
...@@ -156,7 +159,8 @@ class single_sender<SF> { ...@@ -156,7 +159,8 @@ class single_sender<SF> {
SF sf_; SF sf_;
public: public:
using properties = property_set<is_sender<>, is_single<>>; using sender_category = single_sender_tag;
using properties = property_set<>;
constexpr single_sender() = default; constexpr single_sender() = default;
constexpr explicit single_sender(SF sf) : sf_(std::move(sf)) {} constexpr explicit single_sender(SF sf) : sf_(std::move(sf)) {}
...@@ -170,20 +174,18 @@ class single_sender<SF> { ...@@ -170,20 +174,18 @@ class single_sender<SF> {
}; };
template < template <
PUSHMI_TYPE_CONSTRAINT(Sender) Data, PUSHMI_TYPE_CONSTRAINT(SingleSender) Data,
class DSF> class DSF>
class single_sender<Data, DSF> { class single_sender<Data, DSF> {
Data data_; Data data_;
DSF sf_; DSF sf_;
public: public:
using sender_category = single_sender_tag;
using properties = properties_t<Data>; using properties = properties_t<Data>;
static_assert( static_assert(
Sender<Data>, SingleSender<Data>,
"Data must be a sender");
static_assert(
is_single_v<Data>,
"Data must be a single sender"); "Data must be a single sender");
constexpr single_sender() = default; constexpr single_sender() = default;
...@@ -224,13 +226,13 @@ PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn { ...@@ -224,13 +226,13 @@ PUSHMI_INLINE_VAR constexpr struct make_single_sender_fn {
return single_sender<SF>{std::move(sf)}; return single_sender<SF>{std::move(sf)};
} }
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<>&& Sender<Data> && is_single_v<Data>) // (requires True<>&& SingleSender<Data>) //
auto auto
operator()(Data d) const { operator()(Data d) const {
return single_sender<Data, passDSF>{std::move(d)}; return single_sender<Data, passDSF>{std::move(d)};
} }
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data> && is_single_v<Data>) // (requires SingleSender<Data>) //
auto auto
operator()(Data d, DSF sf) const { operator()(Data d, DSF sf) const {
return single_sender<Data, DSF>{std::move(d), std::move(sf)}; return single_sender<Data, DSF>{std::move(d), std::move(sf)};
...@@ -248,12 +250,12 @@ PUSHMI_TEMPLATE(class SF) ...@@ -248,12 +250,12 @@ PUSHMI_TEMPLATE(class SF)
->single_sender<SF>; ->single_sender<SF>;
PUSHMI_TEMPLATE(class Data) PUSHMI_TEMPLATE(class Data)
(requires True<>&& Sender<Data> && is_single_v<Data>) // (requires True<>&& SingleSender<Data>) //
single_sender(Data) single_sender(Data)
->single_sender<Data, passDSF>; ->single_sender<Data, passDSF>;
PUSHMI_TEMPLATE(class Data, class DSF) PUSHMI_TEMPLATE(class Data, class DSF)
(requires Sender<Data> && is_single_v<Data>) // (requires SingleSender<Data>) //
single_sender(Data, DSF) single_sender(Data, DSF)
->single_sender<Data, DSF>; ->single_sender<Data, DSF>;
#endif #endif
......
...@@ -178,14 +178,15 @@ template <class E, class Exec> ...@@ -178,14 +178,15 @@ template <class E, class Exec>
class strand_executor; class strand_executor;
template <class E, class Exec> template <class E, class Exec>
class strand_task { class strand_task
: public single_sender_tag::with_values<any_executor_ref<E>>
::template with_error<E> {
std::shared_ptr<strand_queue<E, Exec>> queue_; std::shared_ptr<strand_queue<E, Exec>> queue_;
public: public:
using properties = property_set< using properties =
is_sender<>, property_set<
property_set_index_t<properties_t<sender_t<Exec>>, is_never_blocking<>>, property_set_index_t<properties_t<sender_t<Exec>>, is_never_blocking<>>>;
is_single<>>;
strand_task(std::shared_ptr<strand_queue<E, Exec>> queue) strand_task(std::shared_ptr<strand_queue<E, Exec>> queue)
: queue_(std::move(queue)) {} : queue_(std::move(queue)) {}
...@@ -200,7 +201,8 @@ class strand_task { ...@@ -200,7 +201,8 @@ class strand_task {
// noone is minding the shop, send a worker // noone is minding the shop, send a worker
guard.unlock(); guard.unlock();
::folly::pushmi::submit( ::folly::pushmi::submit(
::folly::pushmi::schedule(queue_->ex_), strand_queue_receiver<E, Exec>{queue_}); ::folly::pushmi::schedule(queue_->ex_),
strand_queue_receiver<E, Exec>{queue_});
} }
} }
}; };
...@@ -240,12 +242,12 @@ class same_strand_factory_fn { ...@@ -240,12 +242,12 @@ class same_strand_factory_fn {
PUSHMI_TEMPLATE(class E = std::exception_ptr, class Provider) PUSHMI_TEMPLATE(class E = std::exception_ptr, class Provider)
(requires ExecutorProvider<Provider>&& (requires ExecutorProvider<Provider>&&
is_concurrent_sequence_v<executor_t<Provider>>) // is_concurrent_sequence_v<executor_t<Provider>>) //
auto strands(Provider ep) { auto strands(Provider ep) {
return same_strand_factory_fn<E, executor_t<Provider>>{get_executor(ep)}; return same_strand_factory_fn<E, executor_t<Provider>>{get_executor(ep)};
} }
PUSHMI_TEMPLATE(class E = std::exception_ptr, class Exec) PUSHMI_TEMPLATE(class E = std::exception_ptr, class Exec)
(requires Executor<Exec>&& is_concurrent_sequence_v<Exec>) // (requires Executor<Exec>&& is_concurrent_sequence_v<Exec>) //
auto strands(Exec ex) { auto strands(Exec ex) {
return same_strand_factory_fn<E, Exec>{std::move(ex)}; return same_strand_factory_fn<E, Exec>{std::move(ex)};
} }
......
...@@ -29,18 +29,17 @@ template <class... TN> ...@@ -29,18 +29,17 @@ template <class... TN>
struct subject; struct subject;
template <class PS, class... TN> template <class PS, class... TN>
struct subject<PS, TN...> { struct subject<PS, TN...> : single_sender_tag::with_values<TN...> {
using properties = property_set_insert_t< using properties = property_set<>;
property_set<is_sender<>, is_single<>>,
property_set<property_set_index_t<PS, is_single<>>>>;
struct subject_shared { struct subject_shared : single_sender_tag::with_values<TN...> {
using receiver_t = any_receiver<std::exception_ptr, TN...>; using receiver_t = any_receiver<std::exception_ptr, TN...>;
bool done_ = false; bool done_ = false;
::folly::pushmi::detail::opt<std::tuple<std::decay_t<TN>...>> t_; ::folly::pushmi::detail::opt<std::tuple<std::decay_t<TN>...>> t_;
std::exception_ptr ep_; std::exception_ptr ep_;
std::vector<receiver_t> receivers_; std::vector<receiver_t> receivers_;
std::mutex lock_; std::mutex lock_;
PUSHMI_TEMPLATE(class Out) PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, std::exception_ptr> && ReceiveValue<Out, TN...>) (requires ReceiveError<Out, std::exception_ptr> && ReceiveValue<Out, TN...>)
void submit(Out out) { void submit(Out out) {
...@@ -52,8 +51,11 @@ struct subject<PS, TN...> { ...@@ -52,8 +51,11 @@ struct subject<PS, TN...> {
if (!!t_ && done_) { if (!!t_ && done_) {
auto args = *t_; auto args = *t_;
::folly::pushmi::apply( ::folly::pushmi::apply(
::folly::pushmi::set_value, [&out](std::decay_t<TN>&&... ts) {
std::tuple_cat(std::tuple<Out&>{out}, std::move(args))); set_value(out, (std::decay_t<TN>&&) ts...);
},
std::move(args)
);
return; return;
} }
if (done_) { if (done_) {
...@@ -62,28 +64,35 @@ struct subject<PS, TN...> { ...@@ -62,28 +64,35 @@ struct subject<PS, TN...> {
} }
receivers_.push_back(receiver_t{std::move(out)}); receivers_.push_back(receiver_t{std::move(out)});
} }
PUSHMI_TEMPLATE(class... VN) PUSHMI_TEMPLATE(class... VN)
(requires And<SemiMovable<VN>...>) (requires sizeof...(VN) == sizeof...(TN)) //
static constexpr bool AllConvertibleTo() noexcept {
return And<ConvertibleTo<VN, std::decay_t<TN>>...>;
}
PUSHMI_TEMPLATE(class... VN)
(requires subject_shared::AllConvertibleTo<VN...>())
void value(VN&&... vn) { void value(VN&&... vn) {
std::unique_lock<std::mutex> guard(lock_); std::unique_lock<std::mutex> guard(lock_);
t_ = std::tuple<std::decay_t<TN>...>{(VN&&) vn...};
for (auto& out : receivers_) { for (auto& out : receivers_) {
::folly::pushmi::apply( ::folly::pushmi::apply(
::folly::pushmi::set_value, [&out](const std::decay_t<TN>&... ts) { set_value(out, ts...); },
std::tuple<decltype(out), std::decay_t<TN>...>{ *t_
out, detail::as_const(vn)...}); );
} }
t_ = std::make_tuple((VN &&) vn...);
} }
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>) void error(std::exception_ptr e) noexcept {
void error(E e) noexcept {
std::unique_lock<std::mutex> guard(lock_); std::unique_lock<std::mutex> guard(lock_);
ep_ = e; ep_ = e;
for (auto& out : receivers_) { for (auto& out : receivers_) {
set_error(out, std::move(e)); set_error(out, e);
} }
receivers_.clear(); receivers_.clear();
} }
void done() { void done() {
std::unique_lock<std::mutex> guard(lock_); std::unique_lock<std::mutex> guard(lock_);
done_ = true; done_ = true;
......
...@@ -15,18 +15,57 @@ ...@@ -15,18 +15,57 @@
*/ */
#pragma once #pragma once
#include <folly/experimental/pushmi/detail/sender_base.h>
namespace folly { namespace folly {
namespace pushmi { namespace pushmi {
namespace detail {
// inherit: a class that inherits from a bunch of bases
template <class... Ts>
struct inherit : Ts... {
inherit() = default;
constexpr inherit(Ts... ts)
: Ts((Ts&&) ts)...
{}
};
template <class T>
struct inherit<T> : T {
inherit() = default;
explicit constexpr inherit(T t)
: T((T&&) t)
{}
};
template <>
struct inherit<>
{};
} // namespace detail
// derive from this for types that need to find operator|() overloads by ADL // derive from this for types that need to find operator|() overloads by ADL
struct pipeorigin {}; struct pipeorigin {};
// cardinality affects both sender and receiver // traits & tags
struct sender_tag
: sender_base<sender_tag> {
};
namespace detail {
struct virtual_sender_tag
: virtual sender_tag {
};
} // namespace detail
struct cardinality_category {}; struct single_sender_tag
: sender_base<single_sender_tag, detail::virtual_sender_tag> {
};
// flow affects both sender and receiver struct flow_sender_tag
: sender_base<flow_sender_tag, detail::virtual_sender_tag> {
};
struct flow_single_sender_tag
: sender_base<
flow_single_sender_tag,
detail::inherit<flow_sender_tag, single_sender_tag>> {
};
struct flow_category {}; struct flow_category {};
...@@ -34,8 +73,6 @@ struct flow_category {}; ...@@ -34,8 +73,6 @@ struct flow_category {};
struct receiver_category {}; struct receiver_category {};
struct sender_category {};
// blocking affects senders // blocking affects senders
struct blocking_category {}; struct blocking_category {};
...@@ -46,20 +83,6 @@ struct sequence_category {}; ...@@ -46,20 +83,6 @@ struct sequence_category {};
// trait & tag types // trait & tag types
template <class... TN>
struct is_single;
template <>
struct is_single<> {
using property_category = cardinality_category;
};
template <class... TN>
struct is_many;
template <>
struct is_many<> {
using property_category = cardinality_category;
};
template <class... TN> template <class... TN>
struct is_flow; struct is_flow;
template <> template <>
...@@ -74,13 +97,6 @@ struct is_receiver<> { ...@@ -74,13 +97,6 @@ struct is_receiver<> {
using property_category = receiver_category; using property_category = receiver_category;
}; };
template <class... TN>
struct is_sender;
template <>
struct is_sender<> {
using property_category = sender_category;
};
template <class... TN> template <class... TN>
struct is_always_blocking; struct is_always_blocking;
template <> template <>
......
...@@ -28,8 +28,6 @@ using namespace folly::pushmi::aliases; ...@@ -28,8 +28,6 @@ using namespace folly::pushmi::aliases;
#include <folly/portability/GMock.h> #include <folly/portability/GMock.h>
#include <folly/portability/GTest.h> #include <folly/portability/GTest.h>
using namespace testing;
using namespace std::literals; using namespace std::literals;
#if __cpp_deduction_guides >= 201703 && PUSHMI_NOT_ON_WINDOWS #if __cpp_deduction_guides >= 201703 && PUSHMI_NOT_ON_WINDOWS
...@@ -40,6 +38,56 @@ using namespace std::literals; ...@@ -40,6 +38,56 @@ using namespace std::literals;
#define MAKE(x) make_##x #define MAKE(x) make_##x
#endif #endif
template <class...> struct tuple;
template <class...> struct variant;
template <class S>
using values_of_t = mi::sender_values_t<S, tuple, variant>;
template <class S>
using error_of_t = mi::sender_error_t<S, variant>;
using s0 = mi::single_sender_tag::no_values;
static_assert(std::is_same<values_of_t<s0>, variant<>>::value, "");
static_assert(std::is_same<error_of_t<s0>, variant<std::exception_ptr>>::value, "");
using s1 = mi::single_sender_tag::no_values::no_error;
static_assert(std::is_same<values_of_t<s1>, variant<>>::value, "");
static_assert(std::is_same<error_of_t<s1>, variant<>>::value, "");
using s2 = mi::single_sender_tag::with_values<int, int*>;
static_assert(std::is_same<values_of_t<s2>, variant<tuple<int, int*>>>::value, "");
static_assert(std::is_same<error_of_t<s2>, variant<std::exception_ptr>>::value, "");
using s3 = mi::single_sender_tag::with_values<int, int*>::no_error;
static_assert(std::is_same<values_of_t<s3>, variant<tuple<int, int*>>>::value, "");
static_assert(std::is_same<error_of_t<s3>, variant<>>::value, "");
using s4 = mi::single_sender_tag::with_values<int>::with_error<short>;
static_assert(std::is_same<values_of_t<s4>, variant<tuple<int>>>::value, "");
static_assert(std::is_same<error_of_t<s4>, variant<short>>::value, "");
using s5 = mi::single_sender_tag::with_values<int>::or_<char*>;
static_assert(std::is_same<values_of_t<s5>, variant<tuple<int>, tuple<char*>>>::value, "");
static_assert(std::is_same<error_of_t<s5>, variant<std::exception_ptr>>::value, "");
using s6 = mi::single_sender_tag::with_values<int>::or_<char*>::no_error;
static_assert(std::is_same<values_of_t<s6>, variant<tuple<int>, tuple<char*>>>::value, "");
static_assert(std::is_same<error_of_t<s6>, variant<>>::value, "");
using s7 = mi::single_sender_tag::with_values<int>::or_<char*>::with_error<short>;
static_assert(std::is_same<values_of_t<s7>, variant<tuple<int>, tuple<char*>>>::value, "");
static_assert(std::is_same<error_of_t<s7>, variant<short>>::value, "");
using s8 = mi::single_sender_tag::with_values<int>::or_<char*>::with_error<short>::or_<float>;
static_assert(std::is_same<values_of_t<s8>, variant<tuple<int>, tuple<char*>>>::value, "");
static_assert(std::is_same<error_of_t<s8>, variant<short, float>>::value, "");
using s9 = mi::single_sender_tag::with_values<int>::with_error<short>::or_<float>;
static_assert(std::is_same<values_of_t<s9>, variant<tuple<int>>>::value, "");
static_assert(std::is_same<error_of_t<s9>, variant<short, float>>::value, "");
using namespace testing;
void receiver_0_test() { void receiver_0_test() {
auto out0 = mi::MAKE(receiver)(); auto out0 = mi::MAKE(receiver)();
static_assert(mi::Receiver<decltype(out0)>, "out0 not a receiver"); static_assert(mi::Receiver<decltype(out0)>, "out0 not a receiver");
...@@ -216,16 +264,16 @@ void receiver_n_test() { ...@@ -216,16 +264,16 @@ void receiver_n_test() {
void single_sender_test() { void single_sender_test() {
auto in0 = mi::MAKE(single_sender)(); auto in0 = mi::MAKE(single_sender)();
static_assert(mi::Sender<decltype(in0)>, "in0 not a sender"); static_assert(mi::SingleSender<decltype(in0)>, "in0 not a sender");
auto in1 = mi::MAKE(single_sender)(mi::ignoreSF{}); auto in1 = mi::MAKE(single_sender)(mi::ignoreSF{});
static_assert(mi::Sender<decltype(in1)>, "in1 not a sender"); static_assert(mi::SingleSender<decltype(in1)>, "in1 not a sender");
auto in2 = mi::MAKE(single_sender)( auto in2 = mi::MAKE(single_sender)(
[&](auto out) { [&](auto out) {
in0.submit(mi::MAKE(receiver)( in0.submit(mi::MAKE(receiver)(
std::move(out), std::move(out),
mi::on_value([](auto d, int v) { mi::set_value(d, v); }))); mi::on_value([](auto d, int v) { mi::set_value(d, v); })));
}); });
static_assert(mi::Sender<decltype(in2)>, "in2 not a sender"); static_assert(mi::SingleSender<decltype(in2)>, "in2 not a sender");
std::promise<int> p0; std::promise<int> p0;
auto promise0 = mi::MAKE(receiver)(std::move(p0)); auto promise0 = mi::MAKE(receiver)(std::move(p0));
...@@ -404,16 +452,19 @@ void flow_receiver_n_test() { ...@@ -404,16 +452,19 @@ void flow_receiver_n_test() {
void flow_single_sender_test() { void flow_single_sender_test() {
auto in0 = mi::MAKE(flow_single_sender)(); auto in0 = mi::MAKE(flow_single_sender)();
static_assert(mi::Sender<decltype(in0)>, "in0 not a sender"); static_assert(mi::SingleSender<decltype(in0)>, "in0 not a sender");
static_assert(mi::FlowSender<decltype(in0)>, "in0 not flow");
auto in1 = mi::MAKE(flow_single_sender)(mi::ignoreSF{}); auto in1 = mi::MAKE(flow_single_sender)(mi::ignoreSF{});
static_assert(mi::Sender<decltype(in1)>, "in1 not a sender"); static_assert(mi::SingleSender<decltype(in1)>, "in1 not a sender");
static_assert(mi::FlowSender<decltype(in1)>, "in1 not flow");
auto in2 = mi::MAKE(flow_single_sender)( auto in2 = mi::MAKE(flow_single_sender)(
[&](auto out) { [&](auto out) {
in0.submit(mi::MAKE(flow_receiver)( in0.submit(mi::MAKE(flow_receiver)(
std::move(out), std::move(out),
mi::on_value([](auto d, int v) { mi::set_value(d, v); }))); mi::on_value([](auto d, int v) { mi::set_value(d, v); })));
}); });
static_assert(mi::Sender<decltype(in2)>, "in2 not a sender"); static_assert(mi::SingleSender<decltype(in2)>, "in2 not a sender");
static_assert(mi::FlowSender<decltype(in2)>, "in2 not flow");
auto out0 = mi::MAKE(flow_receiver)(); auto out0 = mi::MAKE(flow_receiver)();
auto out1 = mi::MAKE(flow_receiver)( auto out1 = mi::MAKE(flow_receiver)(
...@@ -428,8 +479,10 @@ void flow_single_sender_test() { ...@@ -428,8 +479,10 @@ void flow_single_sender_test() {
void flow_many_sender_test() { void flow_many_sender_test() {
auto in0 = mi::MAKE(flow_many_sender)(); auto in0 = mi::MAKE(flow_many_sender)();
static_assert(mi::Sender<decltype(in0)>, "in0 not a sender"); static_assert(mi::Sender<decltype(in0)>, "in0 not a sender");
static_assert(mi::FlowSender<decltype(in0)>, "in0 not flow");
auto in1 = mi::MAKE(flow_many_sender)(mi::ignoreSF{}); auto in1 = mi::MAKE(flow_many_sender)(mi::ignoreSF{});
static_assert(mi::Sender<decltype(in1)>, "in1 not a sender"); static_assert(mi::Sender<decltype(in1)>, "in1 not a sender");
static_assert(mi::FlowSender<decltype(in1)>, "in1 not flow");
auto in2 = mi::MAKE(flow_many_sender)( auto in2 = mi::MAKE(flow_many_sender)(
[&](auto out) { [&](auto out) {
in0.submit(mi::MAKE(flow_receiver)( in0.submit(mi::MAKE(flow_receiver)(
...@@ -437,6 +490,7 @@ void flow_many_sender_test() { ...@@ -437,6 +490,7 @@ void flow_many_sender_test() {
mi::on_value([](auto d, int v) { mi::set_value(d, v); }))); mi::on_value([](auto d, int v) { mi::set_value(d, v); })));
}); });
static_assert(mi::Sender<decltype(in2)>, "in2 not a sender"); static_assert(mi::Sender<decltype(in2)>, "in2 not a sender");
static_assert(mi::FlowSender<decltype(in2)>, "in2 not flow");
auto out0 = mi::MAKE(flow_receiver)(); auto out0 = mi::MAKE(flow_receiver)();
auto out1 = mi::MAKE(flow_receiver)( auto out1 = mi::MAKE(flow_receiver)(
......
...@@ -39,14 +39,16 @@ TEST(EmptySingleSender, TapAndSubmit) { ...@@ -39,14 +39,16 @@ TEST(EmptySingleSender, TapAndSubmit) {
auto e = op::empty(); auto e = op::empty();
using E = decltype(e); using E = decltype(e);
EXPECT_THAT((v::SenderTo<E, v::any_receiver<>> && v::is_single_v<E>), Eq(true)) EXPECT_THAT(
(v::SenderTo<E, v::any_receiver<>> && v::SingleSender<E>),
Eq(true))
<< "expected empty to return a single sender that can take an any_receiver<>"; << "expected empty to return a single sender that can take an any_receiver<>";
EXPECT_THAT( EXPECT_THAT(
(v::SenderTo< (v::SenderTo<
E, E,
v::any_receiver<std::exception_ptr, int>> && v::any_receiver<std::exception_ptr, int>> &&
v::is_single_v<E>), v::SingleSender<E>),
Eq(true)) Eq(true))
<< "expected empty to return a single sender that can take an any_receiver<int>"; << "expected empty to return a single sender that can take an any_receiver<int>";
...@@ -78,16 +80,14 @@ TEST(JustIntSingleSender, TransformAndSubmit) { ...@@ -78,16 +80,14 @@ TEST(JustIntSingleSender, TransformAndSubmit) {
using J = decltype(j); using J = decltype(j);
EXPECT_THAT( EXPECT_THAT(
(v::SenderTo< (v::SenderTo<J,v::any_receiver<std::exception_ptr, int>> &&
J, v::SingleSender<J>),
v::any_receiver<std::exception_ptr, int>> &&
v::is_single_v<J>),
Eq(true)) Eq(true))
<< "expected empty to return a single sender that can take an any_receiver<int>"; << "expected empty to return a single sender that can take an any_receiver<int>";
int signals = 0; int signals = 0;
int value = 0; int value = 0;
j | std::move(j) |
op::transform( op::transform(
[&](int v) { [&](int v) {
signals += 10000; signals += 10000;
...@@ -131,7 +131,7 @@ TEST(FromIntManySender, TransformAndSubmit) { ...@@ -131,7 +131,7 @@ TEST(FromIntManySender, TransformAndSubmit) {
using M = decltype(m); using M = decltype(m);
EXPECT_THAT( EXPECT_THAT(
(v::SenderTo<M, v::any_receiver<std::exception_ptr, int>> && v::is_many_v<M>), (v::SenderTo<M, v::any_receiver<std::exception_ptr, int>>),
Eq(true)) Eq(true))
<< "expected empty to return a many sender that can take an any_receiver<int>"; << "expected empty to return a many sender that can take an any_receiver<int>";
......
...@@ -442,7 +442,9 @@ class time_source_executor; ...@@ -442,7 +442,9 @@ class time_source_executor;
// //
template <class E, class TP, class NF, class Exec> template <class E, class TP, class NF, class Exec>
class time_source_task { class time_source_task
: public single_sender_tag::with_values<any_time_executor_ref<E, TP>>::
template with_error<E> {
using time_point = std::decay_t<TP>; using time_point = std::decay_t<TP>;
time_point tp_; time_point tp_;
std::shared_ptr<time_source_shared<E, time_point>> source_; std::shared_ptr<time_source_shared<E, time_point>> source_;
......
...@@ -141,7 +141,8 @@ PUSHMI_CONCEPT_DEF( ...@@ -141,7 +141,8 @@ PUSHMI_CONCEPT_DEF(
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
template (class A, class B) template (class A, class B)
concept DerivedFrom, concept DerivedFrom,
__is_base_of(B, A) __is_base_of(B, A) &&
std::is_convertible<const volatile A*, const volatile B*>::value
); );
PUSHMI_CONCEPT_DEF( PUSHMI_CONCEPT_DEF(
......
...@@ -49,11 +49,9 @@ template <class E = std::exception_ptr> ...@@ -49,11 +49,9 @@ template <class E = std::exception_ptr>
class trampoline; class trampoline;
template<class E, class Tag> template<class E, class Tag>
struct trampoline_task { struct trampoline_task
using properties = property_set< : single_sender_tag::with_values<any_executor_ref<E>>::template with_error<E> {
is_sender<>, using properties = property_set<is_maybe_blocking<>>;
is_maybe_blocking<>,
is_single<>>;
PUSHMI_TEMPLATE(class SingleReceiver) PUSHMI_TEMPLATE(class SingleReceiver)
(requires ReceiveValue< (requires ReceiveValue<
......
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