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

consolidate receivers (#53)

fbshipit-source-id: cd20a757d9ff43d80b74c9b7eb63f485ff206f8b
parent 9f236dcd
......@@ -3,6 +3,8 @@ cmake_minimum_required(VERSION 3.7)
project(pushmi-project CXX)
#
option(PUSHMI_USE_CONCEPTS_EMULATION "Use C++14 Concepts Emulation" ON)
option(PUSHMI_USE_CPP_2A "Use C++2a with concepts emulation" OFF)
option(PUSHMI_USE_CPP_17 "Use C++17 with concepts emulation" OFF)
......@@ -35,16 +37,22 @@ if (PUSHMI_USE_CPP_17)
message("Using c++17!")
target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:GNU>:-std=c++17>
$<$<CXX_COMPILER_ID:AppleClang>:-std=c++17>
$<$<CXX_COMPILER_ID:Clang>:-std=c++17>
)
elseif (PUSHMI_USE_CPP_2A)
message("Using c++2a!")
target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:GNU>:-std=c++2a>
$<$<CXX_COMPILER_ID:AppleClang>:-std=c++2a>
$<$<CXX_COMPILER_ID:Clang>:-std=c++2a>
)
else()
message("Using c++14!")
target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:GNU>:-std=c++14>
$<$<CXX_COMPILER_ID:AppleClang>:-std=c++14>
$<$<CXX_COMPILER_ID:Clang>:-std=c++14>
)
endif()
......@@ -66,7 +74,14 @@ target_compile_options(pushmi INTERFACE
endif(PUSHMI_USE_CONCEPTS_EMULATION)
target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:GNU>:-ftemplate-backtrace-limit=0>)
$<$<CXX_COMPILER_ID:GNU>:-ftemplate-backtrace-limit=0>
$<$<CXX_COMPILER_ID:Clang>:-ftemplate-backtrace-limit=0>
$<$<CXX_COMPILER_ID:AppleClang>:-ftemplate-backtrace-limit=0>)
target_compile_options(pushmi INTERFACE
$<$<CXX_COMPILER_ID:Clang>:-stdlib=libc++>
$<$<CXX_COMPILER_ID:AppleClang>:-stdlib=libc++>)
if (PUSHMI_CONCEPTS)
target_compile_options(pushmi INTERFACE $<$<CXX_COMPILER_ID:GNU>:-fconcepts>)
endif ()
......
......@@ -33,19 +33,16 @@ set(header_files
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/concepts.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/boosters.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/piping.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/none.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_many.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/receiver.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_receiver.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/executor.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/inline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/strand.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/constrained_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/time_source.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_sender.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/many_sender.h"
......
......@@ -7,6 +7,7 @@
#include <pushmi/o/transform.h>
#include <pushmi/o/via.h>
#include <pushmi/strand.h>
using namespace pushmi::aliases;
......@@ -31,14 +32,14 @@ void lisp(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::via([cpu]{return cpu;})(
op::via(mi::strands(cpu))(
op::transform([](auto){ return f(); })(
cpu))));
// f on io - g on cpu
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::via([cpu]{return cpu;})(
op::via(mi::strands(cpu))(
op::transform([](auto){ return f(); })(
io))));
}
......@@ -54,14 +55,14 @@ void sugar(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
cpu |
op::transform([](auto){ return f(); }) |
op::via([cpu]{return cpu;}) |
op::via(mi::strands(cpu)) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
// f on io - g on cpu
io |
op::transform([](auto){ return f(); }) |
op::via([cpu]{return cpu;}) |
op::via(mi::strands(cpu)) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
}
......@@ -79,7 +80,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) {
mi::pipe(
cpu,
op::transform([](auto){ return f(); }),
op::via([cpu]{return cpu;}),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
......@@ -87,7 +88,7 @@ void pipe(CPUExecutor cpu, IOExecutor io) {
mi::pipe(
io,
op::transform([](auto){ return f(); }),
op::via([cpu]{return cpu;}),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
}
......
......@@ -5,6 +5,8 @@
#include <pool.h>
#include <pushmi/strand.h>
#include <pushmi/o/request_via.h>
#include <pushmi/o/tap.h>
......@@ -28,7 +30,7 @@ int main()
auto io = ioPool.executor();
auto cpu = cpuPool.executor();
io_operation(io).via([cpu]{ return cpu; }) |
io_operation(io).via(mi::strands(cpu)) |
op::tap([](int v){ printf("cpu pool processing, %d\n", v); }) |
op::submit();
......
......@@ -78,7 +78,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()]() mutable {
e | op::submit([out = std::move(out), ep = std::current_exception()](auto) mutable {
mi::set_error(out, ep);
});
}
......@@ -99,5 +99,6 @@ int main()
std::cout << "OK" << std::endl;
p.stop();
p.wait();
}
......@@ -23,7 +23,7 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn {
return [func, sb, se, driver, initFunc, selector](auto in){
return make_single_sender(
[in, func, sb, se, driver, initFunc, selector](auto out) mutable {
submit(in, make_single(std::move(out),
submit(in, make_receiver(std::move(out),
[func, sb, se, driver, initFunc, selector](auto& out, auto input) {
driver(initFunc, selector, std::move(input), func, sb, se, std::move(out));
}
......
......@@ -25,7 +25,7 @@ namespace execution = std::experimental::execution;
template<class Executor>
struct pool_executor {
using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
using properties = property_set<is_sender<>, is_executor<>, is_never_blocking<>, is_concurrent_sequence<>, is_single<>>;
using e_t = Executor;
e_t e;
......
......@@ -80,7 +80,7 @@ auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()]() mutable {
e | op::submit([out = std::move(out), ep = std::current_exception()](auto) mutable {
mi::set_error(out, ep);
});
}
......
......@@ -33,9 +33,9 @@ auto concat =
[](auto in){
return mi::make_single_sender(
[in](auto out) mutable {
::pushmi::submit(in, mi::make_single(out,
::pushmi::submit(in, mi::make_receiver(out,
[](auto out, auto v){
::pushmi::submit(v, mi::any_single<T, E>(out));
::pushmi::submit(v, mi::any_receiver<E, T>(out));
}));
});
};
......@@ -54,9 +54,9 @@ int main()
op::just(42) |
op::transform([](int i) {
if (i < 42) {
return mi::any_single_sender<std::string>{op::empty<std::string>()};
return mi::any_single_sender<std::exception_ptr, std::string>{op::empty<std::string>()};
}
return mi::any_single_sender<std::string>{op::just(std::to_string(i))};
return mi::any_single_sender<std::exception_ptr, std::string>{op::just(std::to_string(i))};
}) |
concat<std::string> |
op::submit(println);
......
......@@ -18,28 +18,29 @@ auto concat =
[](auto in){
return mi::make_single_sender(
[in](auto out) mutable {
::pushmi::submit(in, mi::make_single(out,
::pushmi::submit(in, mi::make_receiver(out,
[](auto out, auto v){
::pushmi::submit(v, mi::any_single<T, E>(out));
::pushmi::submit(v, mi::any_receiver<E, T>(out));
}));
});
};
int main()
{
auto stop_abort = mi::on_error([](auto) noexcept {});
// support all error value types
op::error(std::exception_ptr{}) |
op::submit();
op::submit(stop_abort);
op::error(std::errc::argument_list_too_long) |
op::submit();
op::submit(stop_abort);
// transform an error
op::error(std::errc::argument_list_too_long) |
op::switch_on_error([](auto e) noexcept { return op::error(std::exception_ptr{}); }) |
op::submit();
op::submit(stop_abort);
// use default value if an error occurs
......@@ -63,7 +64,7 @@ int main()
op::just(42) |
op::transform([](auto v) {
using r_t = mi::any_single_sender<int>;
using r_t = mi::any_single_sender<std::exception_ptr, int>;
if (v < 40) {
return r_t{op::error<int>(std::exception_ptr{})};
} else {
......
......@@ -13,8 +13,7 @@
#include <pool.h>
#include <pushmi/sender.h>
#include <pushmi/single_sender.h>
#include <pushmi/strand.h>
#include <pushmi/o/just.h>
#include <pushmi/o/via.h>
#include <pushmi/o/transform.h>
......@@ -75,7 +74,7 @@ namespace p1055 {
template<class Executor, class Function, class Future>
auto then_execute(Executor&& e, Function&& f, Future&& pred) {
return pred | op::via([e](){return e;}) | op::transform([f](auto v){return f(v);});
return pred | op::via(mi::strands(e)) | op::transform([f](auto v){return f(v);});
}
......@@ -96,7 +95,9 @@ int main()
p1055::then_execute(p.executor(), [](int v){return v*2;}, op::just(21)) | op::get<int>;
sp.stop();
sp.wait();
p.stop();
p.wait();
std::cout << "OK" << std::endl;
......
......@@ -12,8 +12,6 @@
#include <pool.h>
#include <pushmi/sender.h>
#include <pushmi/single_sender.h>
#include <pushmi/o/transform.h>
using namespace pushmi::aliases;
......@@ -65,7 +63,9 @@ int main()
p1055::twoway_execute(p.executor(), [](){return 42;}) | op::get<int>;
sp.stop();
sp.wait();
p.stop();
p.wait();
std::cout << "OK" << std::endl;
......
This diff is collapsed.
......@@ -30,19 +30,10 @@ template<template <class...> class T>
struct construct_deduced;
template<>
struct construct_deduced<none>;
struct construct_deduced<receiver>;
template<>
struct construct_deduced<single>;
template<>
struct construct_deduced<many>;
template<>
struct construct_deduced<flow_single>;
template<>
struct construct_deduced<flow_many>;
struct construct_deduced<flow_receiver>;
template<>
struct construct_deduced<sender>;
......@@ -69,7 +60,9 @@ template <template <class...> class T, class... AN>
using deduced_type_t = pushmi::invoke_result_t<construct_deduced<T>, AN...>;
struct ignoreVF {
void operator()(detail::any) {}
PUSHMI_TEMPLATE(class... VN)
(requires And<ConvertibleTo<VN&&, detail::any>...>)
void operator()(VN&&...) {}
};
struct abortEF {
......@@ -110,18 +103,18 @@ struct priorityZeroF {
};
struct passDVF {
PUSHMI_TEMPLATE(class V, class Data)
PUSHMI_TEMPLATE(class Data, class... VN)
(requires requires (
::pushmi::set_value(std::declval<Data&>(), std::declval<V>())
::pushmi::set_value(std::declval<Data&>(), std::declval<VN>()...)
) && Receiver<Data>)
void operator()(Data& out, V&& v) const {
::pushmi::set_value(out, (V&&) v);
void operator()(Data& out, VN&&... vn) const {
::pushmi::set_value(out, (VN&&) vn...);
}
};
struct passDEF {
PUSHMI_TEMPLATE(class E, class Data)
(requires NoneReceiver<Data, E>)
(requires ReceiveError<Data, E>)
void operator()(Data& out, E e) const noexcept {
::pushmi::set_error(out, e);
}
......@@ -135,16 +128,6 @@ struct passDDF {
}
};
struct passDNXF {
PUSHMI_TEMPLATE(class V, class Data)
(requires requires (
::pushmi::set_next(std::declval<Data&>(), std::declval<V>())
) && Receiver<Data>)
void operator()(Data& out, V&& v) const {
::pushmi::set_next(out, (V&&) v);
}
};
struct passDStrtF {
PUSHMI_TEMPLATE(class Up, class Data)
(requires requires (
......@@ -292,17 +275,6 @@ auto on_done(Fn fn) -> on_done_fn<Fn> {
return on_done_fn<Fn>{std::move(fn)};
}
template <class... Fns>
struct on_next_fn : overload_fn<Fns...> {
constexpr on_next_fn() = default;
using overload_fn<Fns...>::overload_fn;
};
template <class... Fns>
auto on_next(Fns... fns) -> on_next_fn<Fns...> {
return on_next_fn<Fns...>{std::move(fns)...};
}
template <class... Fns>
struct on_starting_fn : overload_fn<Fns...> {
constexpr on_starting_fn() = default;
......
......@@ -17,6 +17,17 @@ namespace pushmi {
struct cardinality_category {};
// Trait
template<class PS>
struct has_cardinality : category_query<PS, cardinality_category> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool has_cardinality_v = has_cardinality<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept Cardinality,
has_cardinality_v<PS>
);
// flow affects both sender and receiver
struct flow_category {};
......@@ -33,47 +44,20 @@ struct executor_category {};
// time and constrained are mutually exclusive refinements of sender (time is a special case of constrained and may be folded in later)
// blocking affects senders
// Silent trait and tag
template<class... TN>
struct is_silent;
// Tag
template<>
struct is_silent<> { using property_category = cardinality_category; };
// Trait
template<class PS>
struct is_silent<PS> : property_query<PS, is_silent<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_silent_v = is_silent<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept Silent,
is_silent_v<PS>
);
struct blocking_category {};
// None trait and tag
template<class... TN>
struct is_none;
// Tag
template<>
struct is_none<> : is_silent<> {};
// Trait
template<class PS>
struct is_none<PS> : property_query<PS, is_none<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_none_v = is_none<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept None,
Silent<PS> && is_none_v<PS>
);
// sequence affects senders
struct sequence_category {};
// Single trait and tag
template<class... TN>
struct is_single;
// Tag
template<>
struct is_single<> : is_none<> {};
struct is_single<> { using property_category = cardinality_category; };
// Trait
template<class PS>
struct is_single<PS> : property_query<PS, is_single<>> {};
......@@ -82,7 +66,7 @@ PUSHMI_INLINE_VAR constexpr bool is_single_v = is_single<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept Single,
None<PS> && is_single_v<PS>
is_single_v<PS>
);
// Many trait and tag
......@@ -90,7 +74,7 @@ template<class... TN>
struct is_many;
// Tag
template<>
struct is_many<> : is_none<> {}; // many::value() does not terminate, so it is not a refinement of single
struct is_many<> { using property_category = cardinality_category; }; // many::value() does not terminate, so it is not a refinement of single
// Trait
template<class PS>
struct is_many<PS> : property_query<PS, is_many<>> {};
......@@ -99,7 +83,7 @@ PUSHMI_INLINE_VAR constexpr bool is_many_v = is_many<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept Many,
None<PS> && is_many_v<PS>
is_many_v<PS>
);
// Flow trait and tag
......@@ -204,57 +188,127 @@ PUSHMI_CONCEPT_DEF(
is_time_v<PS> && is_constrained_v<PS> && is_sender_v<PS>
);
// AlwaysBlocking trait and tag
template<class... TN>
struct is_always_blocking;
// Tag
template<>
struct is_always_blocking<> { using property_category = blocking_category; };
// Trait
template<class PS>
struct is_always_blocking<PS> : property_query<PS, is_always_blocking<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_always_blocking_v = is_always_blocking<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class S, class... PropertyN)
(concept Receiver)(S, PropertyN...),
requires (S& s) (
::pushmi::set_done(s)
) &&
SemiMovable<S> &&
property_query_v<S, PropertyN...> &&
is_receiver_v<S> &&
!is_sender_v<S>
template (class PS)
concept AlwaysBlocking,
is_always_blocking_v<PS> && is_sender_v<PS>
);
// NeverBlocking trait and tag
template<class... TN>
struct is_never_blocking;
// Tag
template<>
struct is_never_blocking<> { using property_category = blocking_category; };
// Trait
template<class PS>
struct is_never_blocking<PS> : property_query<PS, is_never_blocking<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_never_blocking_v = is_never_blocking<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept NeverBlocking,
is_never_blocking_v<PS> && is_sender_v<PS>
);
// MaybeBlocking trait and tag
template<class... TN>
struct is_maybe_blocking;
// Tag
template<>
struct is_maybe_blocking<> { using property_category = blocking_category; };
// Trait
template<class PS>
struct is_maybe_blocking<PS> : property_query<PS, is_maybe_blocking<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_maybe_blocking_v = is_maybe_blocking<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept MaybeBlocking,
is_maybe_blocking_v<PS> && is_sender_v<PS>
);
// FifoSequence trait and tag
template<class... TN>
struct is_fifo_sequence;
// Tag
template<>
struct is_fifo_sequence<> { using property_category = sequence_category; };
// Trait
template<class PS>
struct is_fifo_sequence<PS> : property_query<PS, is_fifo_sequence<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_fifo_sequence_v = is_fifo_sequence<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept FifoSequence,
is_fifo_sequence_v<PS> && is_sender_v<PS>
);
// ConcurrentSequence trait and tag
template<class... TN>
struct is_concurrent_sequence;
// Tag
template<>
struct is_concurrent_sequence<> { using property_category = sequence_category; };
// Trait
template<class PS>
struct is_concurrent_sequence<PS> : property_query<PS, is_concurrent_sequence<>> {};
template<class PS>
PUSHMI_INLINE_VAR constexpr bool is_concurrent_sequence_v = is_concurrent_sequence<PS>::value;
PUSHMI_CONCEPT_DEF(
template (class PS)
concept ConcurrentSequence,
is_concurrent_sequence_v<PS> && is_sender_v<PS>
);
PUSHMI_CONCEPT_DEF(
template (class N, class E = std::exception_ptr)
(concept NoneReceiver)(N, E),
requires(N& n, E&& e) (
::pushmi::set_error(n, (E &&) e)
template (class R, class... PropertyN)
(concept Receiver)(R, PropertyN...),
requires (R& r) (
::pushmi::set_done(r),
::pushmi::set_error(r, std::exception_ptr{})
) &&
Receiver<N> &&
None<N> &&
SemiMovable<E>
SemiMovable<R> &&
property_query_v<R, PropertyN...> &&
is_receiver_v<R> &&
!is_sender_v<R>
);
PUSHMI_CONCEPT_DEF(
template (class S, class T, class E = std::exception_ptr)
(concept SingleReceiver)(S, T, E),
requires(S& s, T&& t) (
::pushmi::set_value(s, (T &&) t) // Semantics: called exactly once.
template (class R, class... VN)
(concept ReceiveValue)(R, VN...),
requires(R& r) (
::pushmi::set_value(r, std::declval<VN &&>()...)
) &&
NoneReceiver<S, E> &&
SemiMovable<T> &&
SemiMovable<E> &&
Single<S>
Receiver<R> &&
// GCC w/-fconcepts ICE on SemiMovable<VN>...
True<> // And<SemiMovable<VN>...>
);
PUSHMI_CONCEPT_DEF(
template (class S, class T, class E = std::exception_ptr)
(concept ManyReceiver)(S, T, E),
requires(S& s, T&& t) (
::pushmi::set_next(s, (T &&) t) // Semantics: called 0-N times.
template (class R, class E = std::exception_ptr)
(concept ReceiveError)(R, E),
requires(R& r, E&& e) (
::pushmi::set_error(r, (E &&) e)
) &&
NoneReceiver<S, E> &&
SemiMovable<T> &&
SemiMovable<E> &&
Many<S>
Receiver<R> &&
SemiMovable<E>
);
// silent does not really make sense, but cannot test for
// None without the error type, use is_none<> to strengthen
// requirements
PUSHMI_CONCEPT_DEF(
template (class D, class... PropertyN)
(concept Sender)(D, PropertyN...),
......@@ -263,7 +317,7 @@ PUSHMI_CONCEPT_DEF(
requires_<Executor<decltype(::pushmi::executor(d))>>
) &&
SemiMovable<D> &&
None<D> &&
Cardinality<D> &&
property_query_v<D, PropertyN...> &&
is_sender_v<D> &&
!is_receiver_v<D>
......@@ -297,47 +351,26 @@ PUSHMI_CONCEPT_DEF(
);
PUSHMI_CONCEPT_DEF(
template (
class N,
class Up,
class PE = std::exception_ptr,
class E = PE)
(concept FlowNoneReceiver)(N, Up, PE, E),
requires(N& n, Up&& up) (
::pushmi::set_starting(n, (Up &&) up)
) &&
FlowReceiver<N> &&
Receiver<Up> &&
SemiMovable<PE> &&
SemiMovable<E> &&
NoneReceiver<Up, PE> &&
NoneReceiver<N, E>
template (class R, class... VN)
(concept FlowReceiveValue)(R, VN...),
Flow<R> &&
ReceiveValue<R, VN...>
);
PUSHMI_CONCEPT_DEF(
template (
class S,
class Up,
class T,
class PE = std::exception_ptr,
class E = PE)
(concept FlowSingleReceiver)(S, Up, T, PE, E),
SingleReceiver<S, T, E> &&
FlowNoneReceiver<S, Up, PE, E>
template (class R, class E = std::exception_ptr)
(concept FlowReceiveError)(R, E),
Flow<R> &&
ReceiveError<R, E>
);
PUSHMI_CONCEPT_DEF(
template (
class S,
class Up,
class T,
class PT = std::ptrdiff_t,
class PE = std::exception_ptr,
class E = PE)
(concept FlowManyReceiver)(S, Up, T, PT, PE, E),
ManyReceiver<S, T, E> &&
ManyReceiver<Up, PT, PE> &&
FlowNoneReceiver<S, Up, PE, E>
template (class R, class Up)
(concept FlowUpTo)(R, Up),
requires(R& r, Up&& up) (
::pushmi::set_starting(r, (Up &&) up)
) &&
Flow<R>
);
PUSHMI_CONCEPT_DEF(
......@@ -374,8 +407,7 @@ PUSHMI_CONCEPT_DEF(
) &&
Sender<D> &&
property_query_v<D, PropertyN...> &&
Constrained<D> &&
None<D>
Constrained<D>
);
PUSHMI_CONCEPT_DEF(
......
......@@ -4,13 +4,13 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "single.h"
#include "receiver.h"
#include "executor.h"
#include "inline.h"
namespace pushmi {
template <class V, class E, class CV>
template <class E, class CV, class... VN>
class any_constrained_single_sender {
union data {
void* pobj_ = nullptr;
......@@ -25,11 +25,11 @@ class any_constrained_single_sender {
static void s_op(data&, data*) {}
static CV s_top(data&) { return CV{}; }
static any_constrained_executor<E, CV> s_executor(data&) { return {}; }
static void s_submit(data&, CV, single<V, E>) {}
static void s_submit(data&, CV, any_receiver<E, VN...>) {}
void (*op_)(data&, data*) = vtable::s_op;
CV (*top_)(data&) = vtable::s_top;
any_constrained_executor<E, CV> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, CV, single<V, E>) = vtable::s_submit;
void (*submit_)(data&, CV, any_receiver<E, VN...>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_;
......@@ -48,7 +48,7 @@ class any_constrained_single_sender {
static any_constrained_executor<E, CV> executor(data& src) {
return any_constrained_executor<E, CV>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, CV at, single<V, E> out) {
static void submit(data& src, CV at, any_receiver<E, VN...> out) {
::pushmi::submit(
*static_cast<Wrapped*>(src.pobj_),
std::move(at),
......@@ -75,7 +75,7 @@ class any_constrained_single_sender {
static any_constrained_executor<E, CV> executor(data& src) {
return any_constrained_executor<E, CV>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, CV cv, single<V, E> out) {
static void submit(data& src, CV cv, any_receiver<E, VN...> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_),
std::move(cv),
......@@ -100,7 +100,7 @@ class any_constrained_single_sender {
std::swap(that.vptr_, vptr_);
}
PUSHMI_TEMPLATE (class Wrapped)
(requires ConstrainedSenderTo<wrapped_t<Wrapped>, single<V, E>>)
(requires ConstrainedSenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>>)
explicit any_constrained_single_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: any_constrained_single_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {
}
......@@ -118,15 +118,15 @@ class any_constrained_single_sender {
any_constrained_executor<E, CV> executor() {
return vptr_->executor_(data_);
}
void submit(CV at, single<V, E> out) {
void submit(CV at, any_receiver<E, VN...> out) {
vptr_->submit_(data_, std::move(at), std::move(out));
}
};
// Class static definitions:
template <class V, class E, class CV>
constexpr typename any_constrained_single_sender<V, E, CV>::vtable const
any_constrained_single_sender<V, E, CV>::noop_;
template <class E, class CV, class... VN>
constexpr typename any_constrained_single_sender<E, CV, VN...>::vtable const
any_constrained_single_sender<E, CV, VN...>::noop_;
template<class SF, class ZF, class EXF>
// (requires Invocable<ZF&> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
......@@ -151,7 +151,7 @@ class constrained_single_sender<SF, ZF, EXF> {
}
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class CV, class Out)
(requires Regular<CV> && Receiver<Out, is_single<>> &&
(requires Regular<CV> && Receiver<Out> &&
Invocable<SF&, CV, Out>)
void submit(CV cv, Out out) {
sf_(std::move(cv), std::move(out));
......@@ -184,7 +184,7 @@ class constrained_single_sender<Data, DSF, DZF, DEXF> {
}
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class CV, class Out)
(requires Regular<CV> && Receiver<Out, is_single<>> &&
(requires Regular<CV> && Receiver<Out> &&
Invocable<DSF&, Data&, CV, Out>)
void submit(CV cv, Out out) {
sf_(data_, std::move(cv), std::move(out));
......@@ -277,13 +277,4 @@ template<>
struct construct_deduced<constrained_single_sender>
: make_constrained_single_sender_fn {};
// template <
// class V,
// class E = std::exception_ptr,
// class CV = std::chrono::system_clock::time_point,
// ConstrainedSenderTo<single<V, E>, is_single<>> Wrapped>
// auto erase_cast(Wrapped w) {
// return constrained_single_sender<V, E>{std::move(w)};
// }
} // namespace pushmi
......@@ -13,7 +13,7 @@
// disable buggy compatibility warning about "requires" and "concept" being
// C++20 keywords.
#if defined(__clang__)
#if defined(__clang__) && not defined(__APPLE__)
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_BEGIN \
_Pragma("GCC diagnostic push") \
_Pragma("GCC diagnostic ignored \"-Wunknown-pragmas\"") \
......@@ -23,7 +23,17 @@
/**/
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_END \
_Pragma("GCC diagnostic pop")
#elif defined(__clang__) && defined(__APPLE__)
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_BEGIN \
_Pragma("GCC diagnostic push") \
_Pragma("GCC diagnostic ignored \"-Wunknown-pragmas\"") \
_Pragma("GCC diagnostic ignored \"-Wpragmas\"") \
_Pragma("GCC diagnostic ignored \"-Wfloat-equal\"") \
/**/
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_END \
_Pragma("GCC diagnostic pop")
#else
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_BEGIN
#define PUSHMI_PP_IGNORE_CXX2A_COMPAT_END
// #pragma GCC diagnostic push
......
......@@ -26,15 +26,10 @@ PUSHMI_TEMPLATE (class S, class E)
void set_error(S& s, E e) noexcept(noexcept(s.error(std::move(e)))) {
s.error(std::move(e));
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>().value(std::declval<V&&>())))
void set_value(S& s, V&& v) noexcept(noexcept(s.value((V&&) v))) {
s.value((V&&) v);
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>().next(std::declval<V&&>())))
void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
s.next((V&&) v);
PUSHMI_TEMPLATE (class S, class... VN)
(requires requires (std::declval<S&>().value(std::declval<VN&&>()...)))
void set_value(S& s, VN&&... vn) noexcept(noexcept(s.value((VN&&) vn...))) {
s.value((VN&&) vn...);
}
PUSHMI_TEMPLATE (class S, class Up)
......@@ -50,7 +45,9 @@ auto executor(SD& sd) noexcept(noexcept(sd.executor())) {
}
PUSHMI_TEMPLATE (class SD, class Out)
(requires requires (std::declval<SD&>().submit(std::declval<Out>())))
(requires requires (
std::declval<SD&>().submit(std::declval<Out>())
))
void submit(SD& sd, Out out) noexcept(noexcept(sd.submit(std::move(out)))) {
sd.submit(std::move(out));
}
......@@ -64,7 +61,7 @@ auto top(SD& sd) noexcept(noexcept(sd.top())) {
PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires (
std::declval<SD&>().submit(
std::declval<TP(&)(TP)>()(std::declval<SD&>().top()),
std::declval<TP(&)(TP)>()(top(std::declval<SD&>())),
std::declval<Out>())
))
void submit(SD& sd, TP tp, Out out)
......@@ -86,15 +83,10 @@ PUSHMI_TEMPLATE (class S, class E)
void set_error(S& s, E e) noexcept(noexcept(s->error(std::move(e)))) {
s->error(std::move(e));
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>()->value(std::declval<V&&>())))
void set_value(S& s, V&& v) noexcept(noexcept(s->value((V&&) v))) {
s->value((V&&) v);
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires (std::declval<S&>()->next(std::declval<V&&>())))
void set_next(S& s, V&& v) noexcept(noexcept(s->next((V&&) v))) {
s->next((V&&) v);
PUSHMI_TEMPLATE (class S, class... VN)
(requires requires (std::declval<S&>()->value(std::declval<VN&&>()...)))
void set_value(S& s, VN&&... vn) noexcept(noexcept(s->value((VN&&) vn...))) {
s->value((VN&&) vn...);
}
PUSHMI_TEMPLATE (class S, class Up)
......@@ -124,7 +116,7 @@ auto top(SD& sd) noexcept(noexcept(sd->top())) {
PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires (
std::declval<SD&>()->submit(
std::declval<TP(&)(TP)>()(std::declval<SD&>()->top()),
std::declval<TP(&)(TP)>()(top(std::declval<SD&>())),
std::declval<Out>())
))
void submit(SD& sd, TP tp, Out out)
......@@ -136,26 +128,25 @@ void submit(SD& sd, TP tp, Out out)
// add support for std::promise externally
//
// std::promise does not support the done signal.
// either set_value or set_error must be called
template <class T>
void set_done(std::promise<T>& p) noexcept(
noexcept(p.set_exception(std::make_exception_ptr(0)))) {
p.set_exception(std::make_exception_ptr(
std::logic_error("std::promise does not support done.")));
}
inline void set_done(std::promise<void>& p) noexcept(noexcept(p.set_value())) {
p.set_value();
}
void set_done(std::promise<T>& p) noexcept {}
template <class T>
void set_error(std::promise<T>& s, std::exception_ptr e) noexcept {
s.set_exception(std::move(e));
void set_error(std::promise<T>& p, std::exception_ptr e) noexcept {
p.set_exception(std::move(e));
}
template <class T, class E>
void set_error(std::promise<T>& s, E e) noexcept {
s.set_exception(std::make_exception_ptr(std::move(e)));
void set_error(std::promise<T>& p, E e) noexcept {
p.set_exception(std::make_exception_ptr(std::move(e)));
}
template <class T>
void set_value(std::promise<T>& s, T t) {
s.set_value(std::move(t));
void set_value(std::promise<T>& p, T t) noexcept(noexcept(p.set_value(std::move(t)))) {
p.set_value(std::move(t));
}
inline void set_value(std::promise<void>& p) noexcept(noexcept(p.set_value())) {
p.set_value();
}
//
......@@ -173,17 +164,11 @@ PUSHMI_TEMPLATE (class S, class E)
void set_error(std::reference_wrapper<S> s, E e) noexcept {
set_error(s.get(), std::move(e));
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires ( set_value(std::declval<S&>(), std::declval<V&&>()) ))
void set_value(std::reference_wrapper<S> s, V&& v) noexcept(
noexcept(set_value(s.get(), (V&&) v))) {
set_value(s.get(), (V&&) v);
}
PUSHMI_TEMPLATE (class S, class V)
(requires requires ( set_next(std::declval<S&>(), std::declval<V&&>()) ))
void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
noexcept(set_next(s.get(), (V&&) v))) {
set_next(s.get(), (V&&) v);
PUSHMI_TEMPLATE (class S, class... VN)
(requires requires ( set_value(std::declval<S&>(), std::declval<VN&&>()...) ))
void set_value(std::reference_wrapper<S> s, VN&&... vn) noexcept(
noexcept(set_value(s.get(), (VN&&) vn...))) {
set_value(s.get(), (VN&&) vn...);
}
PUSHMI_TEMPLATE (class S, class Up)
(requires requires ( set_starting(std::declval<S&>(), std::declval<Up&&>()) ))
......@@ -202,6 +187,7 @@ void submit(std::reference_wrapper<SD> sd, Out out) noexcept(
noexcept(submit(sd.get(), std::move(out)))) {
submit(sd.get(), std::move(out));
}
PUSHMI_TEMPLATE (class SD)
(requires requires ( top(std::declval<SD&>()) ))
auto top(std::reference_wrapper<SD> sd) noexcept(noexcept(top(sd.get()))) {
......@@ -248,30 +234,15 @@ struct set_error_fn {
}
};
struct set_value_fn {
PUSHMI_TEMPLATE (class S, class V)
PUSHMI_TEMPLATE (class S, class... VN)
(requires requires (
set_value(std::declval<S&>(), std::declval<V&&>()),
set_value(std::declval<S&>(), std::declval<VN&&>()...),
set_error(std::declval<S&>(), std::current_exception())
))
void operator()(S&& s, V&& v) const
noexcept(noexcept(set_value(s, (V&&) v))) {
void operator()(S&& s, VN&&... vn) const
noexcept(noexcept(set_value(s, (VN&&) vn...))) {
try {
set_value(s, (V&&) v);
} catch (...) {
set_error(s, std::current_exception());
}
}
};
struct set_next_fn {
PUSHMI_TEMPLATE (class S, class V)
(requires requires (
set_next(std::declval<S&>(), std::declval<V&&>()),
set_error(std::declval<S&>(), std::current_exception())
))
void operator()(S&& s, V&& v) const
noexcept(noexcept(set_next(s, (V&&) v))) {
try {
set_next(s, (V&&) v);
set_value(s, (VN&&) vn...);
} catch (...) {
set_error(s, std::current_exception());
}
......@@ -314,6 +285,15 @@ struct do_submit_fn {
submit(s, std::move(out));
}
PUSHMI_TEMPLATE (class SD, class Out)
(requires requires (
submit(std::declval<SD&>(), top(std::declval<SD&>()), std::declval<Out>())
))
void operator()(SD&& s, Out out) const
noexcept(noexcept(submit(s, top(s), std::move(out)))) {
submit(s, top(s), std::move(out));
}
PUSHMI_TEMPLATE (class SD, class TP, class Out)
(requires requires (
submit(
......@@ -342,7 +322,6 @@ struct get_top_fn {
PUSHMI_INLINE_VAR constexpr __adl::set_done_fn set_done{};
PUSHMI_INLINE_VAR constexpr __adl::set_error_fn set_error{};
PUSHMI_INLINE_VAR constexpr __adl::set_value_fn set_value{};
PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{};
PUSHMI_INLINE_VAR constexpr __adl::set_starting_fn set_starting{};
PUSHMI_INLINE_VAR constexpr __adl::get_executor_fn executor{};
PUSHMI_INLINE_VAR constexpr __adl::do_submit_fn submit{};
......@@ -351,11 +330,11 @@ PUSHMI_INLINE_VAR constexpr __adl::get_top_fn top{};
template <class T>
struct property_set_traits<std::promise<T>> {
using properties = property_set<is_receiver<>, is_single<>>;
using properties = property_set<is_receiver<>>;
};
template <>
struct property_set_traits<std::promise<void>> {
using properties = property_set<is_receiver<>, is_none<>>;
using properties = property_set<is_receiver<>>;
};
} // namespace pushmi
This diff is collapsed.
......@@ -4,17 +4,17 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "flow_many.h"
#include "flow_receiver.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi {
template <class V, class PV, class PE, class E>
class flow_many_sender<V, PV, PE, E> {
template <class PE, class PV, class E, class... VN>
class any_flow_many_sender {
union data {
void* pobj_ = nullptr;
char buffer_[sizeof(V)]; // can hold a V in-situ
char buffer_[sizeof(std::tuple<VN...>)]; // can hold a V in-situ
} data_{};
template <class Wrapped>
static constexpr bool insitu() {
......@@ -24,15 +24,15 @@ class flow_many_sender<V, PV, PE, E> {
struct vtable {
static void s_op(data&, data*) {}
static any_executor<E> s_executor(data&) { return {}; }
static void s_submit(data&, any_flow_many<V, PV, PE, E>) {}
static void s_submit(data&, any_flow_receiver<PE, PV, E, VN...>) {}
void (*op_)(data&, data*) = vtable::s_op;
any_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, any_flow_many<V, PV, PE, E>) = vtable::s_submit;
void (*submit_)(data&, any_flow_receiver<PE, PV, E, VN...>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_;
template <class Wrapped>
flow_many_sender(Wrapped obj, std::false_type) : flow_many_sender() {
any_flow_many_sender(Wrapped obj, std::false_type) : any_flow_many_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
......@@ -42,7 +42,7 @@ class flow_many_sender<V, PV, PE, E> {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
static void submit(data& src, any_flow_receiver<PE, PV, E, VN...> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
......@@ -51,8 +51,8 @@ class flow_many_sender<V, PV, PE, E> {
vptr_ = &vtbl;
}
template <class Wrapped>
flow_many_sender(Wrapped obj, std::true_type) noexcept
: flow_many_sender() {
any_flow_many_sender(Wrapped obj, std::true_type) noexcept
: any_flow_many_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
......@@ -63,7 +63,7 @@ class flow_many_sender<V, PV, PE, E> {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, any_flow_many<V, PV, PE, E> out) {
static void submit(data& src, any_flow_receiver<PE, PV, E, VN...> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_),
std::move(out));
......@@ -75,40 +75,40 @@ class flow_many_sender<V, PV, PE, E> {
}
template <class T, class U = std::decay_t<T>>
using wrapped_t =
std::enable_if_t<!std::is_same<U, flow_many_sender>::value, U>;
std::enable_if_t<!std::is_same<U, any_flow_many_sender>::value, U>;
public:
using properties = property_set<is_sender<>, is_flow<>, is_many<>>;
flow_many_sender() = default;
flow_many_sender(flow_many_sender&& that) noexcept
: flow_many_sender() {
any_flow_many_sender() = default;
any_flow_many_sender(any_flow_many_sender&& that) noexcept
: any_flow_many_sender() {
that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_);
}
PUSHMI_TEMPLATE (class Wrapped)
(requires FlowSender<wrapped_t<Wrapped>, is_many<>>)
explicit flow_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: flow_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~flow_many_sender() {
explicit any_flow_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: any_flow_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_flow_many_sender() {
vptr_->op_(data_, nullptr);
}
flow_many_sender& operator=(flow_many_sender&& that) noexcept {
this->~flow_many_sender();
new ((void*)this) flow_many_sender(std::move(that));
any_flow_many_sender& operator=(any_flow_many_sender&& that) noexcept {
this->~any_flow_many_sender();
new ((void*)this) any_flow_many_sender(std::move(that));
return *this;
}
any_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(any_flow_many<V, PV, PE, E> out) {
void submit(any_flow_receiver<PE, PV, E, VN...> out) {
vptr_->submit_(data_, std::move(out));
}
};
// Class static definitions:
template <class V, class PV, class PE, class E>
constexpr typename flow_many_sender<V, PV, PE, E>::vtable const
flow_many_sender<V, PV, PE, E>::noop_;
template <class PE, class PV, class E, class... VN>
constexpr typename any_flow_many_sender<PE, PV, E, VN...>::vtable const
any_flow_many_sender<PE, PV, E, VN...>::noop_;
template <class SF, class EXF>
class flow_many_sender<SF, EXF> {
......@@ -126,7 +126,7 @@ class flow_many_sender<SF, EXF> {
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_many<>, is_flow<>> && Invocable<SF&, Out>)
(requires FlowReceiver<Out> && Invocable<SF&, Out>)
void submit(Out out) {
sf_(std::move(out));
}
......@@ -151,7 +151,7 @@ class flow_many_sender<Data, DSF, DEXF> {
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>, is_flow<>> PUSHMI_AND
(requires PUSHMI_EXP(lazy::FlowReceiver<Out> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
......@@ -224,9 +224,6 @@ PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
flow_many_sender(Data, DSF, DEXF) -> flow_many_sender<Data, DSF, DEXF>;
#endif
template <class V, class PV = std::ptrdiff_t, class PE = std::exception_ptr, class E = PE>
using any_flow_many_sender = flow_many_sender<V, PV, PE, E>;
template<>
struct construct_deduced<flow_many_sender>
: make_flow_many_sender_fn {};
......
......@@ -4,17 +4,17 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "flow_single.h"
#include "flow_receiver.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi {
template <class V, class PE = std::exception_ptr, class E = PE>
template <class PE, class E, class... VN>
class any_flow_single_sender {
union data {
void* pobj_ = nullptr;
char buffer_[sizeof(V)]; // can hold a V in-situ
char buffer_[sizeof(std::tuple<VN...>)]; // can hold a V in-situ
} data_{};
template <class Wrapped>
static constexpr bool insitu() {
......@@ -24,10 +24,10 @@ class any_flow_single_sender {
struct vtable {
static void s_op(data&, data*) {}
static any_executor<E> s_executor(data&) { return {}; }
static void s_submit(data&, flow_single<V, PE, E>) {}
static void s_submit(data&, any_flow_receiver<PE, std::ptrdiff_t, E, VN...>) {}
void (*op_)(data&, data*) = vtable::s_op;
any_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, flow_single<V, PE, E>) = vtable::s_submit;
void (*submit_)(data&, any_flow_receiver<PE, std::ptrdiff_t, E, VN...>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_;
......@@ -42,7 +42,7 @@ class any_flow_single_sender {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, flow_single<V, PE, E> out) {
static void submit(data& src, any_flow_receiver<PE, std::ptrdiff_t, E, VN...> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
......@@ -63,7 +63,7 @@ class any_flow_single_sender {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, flow_single<V, PE, E> out) {
static void submit(data& src, any_flow_receiver<PE, std::ptrdiff_t, E, VN...> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_),
std::move(out));
......@@ -100,15 +100,15 @@ class any_flow_single_sender {
any_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(flow_single<V, PE, E> out) {
void submit(any_flow_receiver<PE, std::ptrdiff_t, E, VN...> out) {
vptr_->submit_(data_, std::move(out));
}
};
// Class static definitions:
template <class V, class PE, class E>
constexpr typename any_flow_single_sender<V, PE, E>::vtable const
any_flow_single_sender<V, PE, E>::noop_;
template <class PE, class E, class... VN>
constexpr typename any_flow_single_sender<PE, E, VN...>::vtable const
any_flow_single_sender<PE, E, VN...>::noop_;
template <class SF, class EXF>
class flow_single_sender<SF, EXF> {
......@@ -126,7 +126,7 @@ class flow_single_sender<SF, EXF> {
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_single<>, is_flow<>> && Invocable<SF&, Out>)
(requires Receiver<Out> && Invocable<SF&, Out>)
void submit(Out out) {
sf_(std::move(out));
}
......@@ -151,7 +151,7 @@ class flow_single_sender<Data, DSF, DEXF> {
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_single<>, is_flow<>> PUSHMI_AND
(requires PUSHMI_EXP(lazy::Receiver<Out> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
......@@ -228,10 +228,5 @@ template<>
struct construct_deduced<flow_single_sender>
: make_flow_single_sender_fn {};
// // TODO constrain me
// template <class V, class E = std::exception_ptr, Sender Wrapped>
// auto erase_cast(Wrapped w) {
// return flow_single_sender<V, E>{std::move(w)};
// }
} // namespace pushmi
......@@ -23,10 +23,6 @@ struct property_set;
// trait & tag types
template<class...TN>
struct is_silent;
template<class...TN>
struct is_none;
template<class...TN>
struct is_single;
template<class...TN>
struct is_many;
......@@ -40,24 +36,39 @@ struct is_receiver;
template<class...TN>
struct is_sender;
template<class... TN>
struct is_executor;
template<class...TN>
struct is_time;
template<class...TN>
struct is_constrained;
// implementation types
template<class... TN>
struct is_always_blocking;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class none;
template<class... TN>
struct is_never_blocking;
template<class... TN>
struct is_maybe_blocking;
template<class... TN>
struct is_fifo_sequence;
template<class... TN>
struct is_concurrent_sequence;
// implementation types
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class sender;
class receiver;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class single;
class flow_receiver;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class many;
class sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class single_sender;
......@@ -71,34 +82,57 @@ class constrained_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class time_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_single;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many_sender;
template <
class E = std::exception_ptr,
class... VN>
class any_receiver;
template <
class PE=std::exception_ptr,
class PV=std::ptrdiff_t,
class E=PE,
class... VN>
class any_flow_receiver;
template<
class V,
class E = std::exception_ptr>
class any_single_sender;
class E = std::exception_ptr,
class... VN>
struct any_single_sender;
template<
class E = std::exception_ptr,
class... VN>
struct any_many_sender;
template <
class PE=std::exception_ptr,
class E=PE,
class... VN>
class any_flow_single_sender;
template <
class PE=std::exception_ptr,
class PV=std::ptrdiff_t,
class E=PE,
class... VN>
class any_flow_many_sender;
template<
class V,
class E = std::exception_ptr,
class C = std::ptrdiff_t>
class C = std::ptrdiff_t,
class... VN>
struct any_constrained_single_sender;
template<
class V,
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
class TP = std::chrono::system_clock::time_point,
class... VN>
class any_time_single_sender;
template<
......
......@@ -10,16 +10,17 @@ namespace pushmi {
class inline_constrained_executor_t {
public:
using properties = property_set<is_constrained<>, is_executor<>, is_single<>>;
using properties = property_set<is_constrained<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
std::ptrdiff_t top() {
return 0;
}
auto executor() { return *this; }
PUSHMI_TEMPLATE(class CV, class Out)
(requires Regular<CV> && Receiver<Out, is_single<>>)
(requires Regular<CV> && Receiver<Out>)
void submit(CV, Out out) {
::pushmi::set_value(std::move(out), *this);
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
......@@ -35,17 +36,18 @@ inline inline_constrained_executor_t inline_constrained_executor() {
class inline_time_executor_t {
public:
using properties = property_set<is_time<>, is_executor<>, is_single<>>;
using properties = property_set<is_time<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
auto top() {
return std::chrono::system_clock::now();
}
auto executor() { return *this; }
PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP> && Receiver<Out, is_single<>>)
(requires Regular<TP> && Receiver<Out>)
void submit(TP tp, Out out) {
std::this_thread::sleep_until(tp);
::pushmi::set_value(std::move(out), *this);
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
......@@ -61,13 +63,14 @@ inline inline_time_executor_t inline_time_executor() {
class inline_executor_t {
public:
using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
using properties = property_set<is_sender<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
auto executor() { return *this; }
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_single<>>)
(requires Receiver<Out>)
void submit(Out out) {
::pushmi::set_value(std::move(out), *this);
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
......
This diff is collapsed.
......@@ -4,17 +4,17 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "many.h"
#include "receiver.h"
#include "executor.h"
#include "trampoline.h"
namespace pushmi {
template <class V, class E = std::exception_ptr>
template <class E, class... VN>
class any_many_sender {
union data {
void* pobj_ = nullptr;
char buffer_[sizeof(V)]; // can hold a V in-situ
char buffer_[sizeof(std::tuple<VN...>)]; // can hold a V in-situ
} data_{};
template <class Wrapped>
static constexpr bool insitu() {
......@@ -24,10 +24,10 @@ class any_many_sender {
struct vtable {
static void s_op(data&, data*) {}
static any_executor<E> s_executor(data&) { return {}; }
static void s_submit(data&, many<V, E>) {}
static void s_submit(data&, any_receiver<E, VN...>) {}
void (*op_)(data&, data*) = vtable::s_op;
any_executor<E> (*executor_)(data&) = vtable::s_executor;
void (*submit_)(data&, many<V, E>) = vtable::s_submit;
void (*submit_)(data&, any_receiver<E, VN...>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
vtable const* vptr_ = &noop_;
......@@ -42,7 +42,7 @@ class any_many_sender {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, many<V, E> out) {
static void submit(data& src, any_receiver<E, VN...> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
}
};
......@@ -63,7 +63,7 @@ class any_many_sender {
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, many<V, E> out) {
static void submit(data& src, any_receiver<E, VN...> out) {
::pushmi::submit(
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
}
......@@ -86,7 +86,7 @@ class any_many_sender {
}
PUSHMI_TEMPLATE(class Wrapped)
(requires SenderTo<wrapped_t<Wrapped>, many<V, E>, is_many<>>)
(requires SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>, is_many<>>)
explicit any_many_sender(Wrapped obj) noexcept(insitu<Wrapped>())
: any_many_sender{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_many_sender() {
......@@ -100,15 +100,15 @@ class any_many_sender {
any_executor<E> executor() {
return vptr_->executor_(data_);
}
void submit(many<V, E> out) {
void submit(any_receiver<E, VN...> out) {
vptr_->submit_(data_, std::move(out));
}
};
// Class static definitions:
template <class V, class E>
constexpr typename any_many_sender<V, E>::vtable const
any_many_sender<V, E>::noop_;
template <class E, class... VN>
constexpr typename any_many_sender<E, VN...>::vtable const
any_many_sender<E, VN...>::noop_;
template <class SF, class EXF>
class many_sender<SF, EXF> {
......@@ -126,7 +126,7 @@ class many_sender<SF, EXF> {
auto executor() { return exf_(); }
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND lazy::Invocable<SF&, Out>))
(requires PUSHMI_EXP(lazy::Receiver<Out> PUSHMI_AND lazy::Invocable<SF&, Out>))
void submit(Out out) {
sf_(std::move(out));
}
......@@ -151,7 +151,7 @@ class many_sender<Data, DSF, DEXF> {
auto executor() { return exf_(data_); }
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out, is_many<>> PUSHMI_AND
(requires PUSHMI_EXP(lazy::Receiver<Out> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
......@@ -227,12 +227,5 @@ many_sender(Data, DSF, DEXF) -> many_sender<Data, DSF, DEXF>;
template<>
struct construct_deduced<many_sender> : make_many_sender_fn {};
// template <
// class V,
// class E = std::exception_ptr,
// SenderTo<many<V, E>, is_many<>> Wrapped>
// auto erase_cast(Wrapped w) {
// return many_sender<V, E>{std::move(w)};
// }
} // namespace pushmi
......@@ -13,7 +13,7 @@ namespace pushmi {
//
struct new_thread_executor {
using properties = property_set<is_sender<>, is_executor<>, is_single<>>;
using properties = property_set<is_sender<>, is_executor<>, is_never_blocking<>, is_concurrent_sequence<>, is_single<>>;
new_thread_executor executor() { return {}; }
PUSHMI_TEMPLATE(class Out)
......
This diff is collapsed.
......@@ -6,7 +6,7 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "../single.h"
#include "../receiver.h"
#include "../single_sender.h"
#include "submit.h"
#include "extension_operators.h"
......@@ -20,22 +20,21 @@ private:
template <class F>
struct impl {
F f_;
PUSHMI_TEMPLATE(class Out)
PUSHMI_TEMPLATE(class Data, class Out)
(requires Receiver<Out>)
void operator()(Out out) {
void operator()(Data&, Out out) {
auto sender = f_();
PUSHMI_IF_CONSTEXPR( ((bool)TimeSender<decltype(sender)>) (
::pushmi::submit(sender, ::pushmi::now(id(sender)), std::move(out));
) else (
::pushmi::submit(sender, std::move(out));
));
}
};
public:
PUSHMI_TEMPLATE(class F)
(requires Invocable<F&>)
auto operator()(F f) const {
return make_single_sender(impl<F>{std::move(f)});
struct sender_base : single_sender<> {
using properties = properties_t<invoke_result_t<F&>>;
};
return make_single_sender(sender_base{}, impl<F>{std::move(f)});
}
} defer {};
......
......@@ -6,37 +6,33 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "../sender.h"
#include "../single_sender.h"
#include "../detail/functional.h"
#include "submit.h"
#include "extension_operators.h"
namespace pushmi {
namespace detail {
template <class V>
struct single_empty_impl {
PUSHMI_TEMPLATE(class Out)
(requires SingleReceiver<Out, V>)
void operator()(Out out) {
::pushmi::set_done(out);
}
struct single_empty_sender_base : single_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_single<>, is_always_blocking<>, is_fifo_sequence<>>;
};
struct empty_impl {
template <class... VN>
struct single_empty_impl {
PUSHMI_TEMPLATE(class Out)
(requires NoneReceiver<Out>)
void operator()(Out out) {
(requires ReceiveValue<Out, VN...>)
void operator()(single_empty_sender_base&, Out out) {
::pushmi::set_done(out);
}
};
}
namespace operators {
template <class V>
template <class... VN>
auto empty() {
return make_single_sender(detail::single_empty_impl<V>{});
return make_single_sender(detail::single_empty_sender_base{}, detail::single_empty_impl<VN...>{});
}
inline auto empty() {
return make_sender(detail::empty_impl{});
return make_single_sender(detail::single_empty_sender_base{}, detail::single_empty_impl<>{});
}
} // namespace operators
......
......@@ -6,27 +6,20 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "../sender.h"
#include "submit.h"
#include "extension_operators.h"
namespace pushmi {
namespace detail {
template <class E>
struct error_impl {
E e_;
PUSHMI_TEMPLATE(class Out)
(requires NoneReceiver<Out, E>)
void operator()(Out out) {
::pushmi::set_error(out, std::move(e_));
}
struct single_error_sender_base : single_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_single<>, is_always_blocking<>, is_fifo_sequence<>>;
};
template <class V, class E>
template <class E, class... VN>
struct single_error_impl {
E e_;
PUSHMI_TEMPLATE(class Out)
(requires SingleReceiver<Out, V, E>)
void operator()(Out out) {
(requires ReceiveError<Out, E> && ReceiveValue<Out, VN...>)
void operator()(single_error_sender_base&, Out out) {
::pushmi::set_error(out, std::move(e_));
}
};
......@@ -34,16 +27,10 @@ namespace detail {
namespace operators {
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>)
auto error(E e) {
return make_sender(detail::error_impl<E>{std::move(e)});
}
PUSHMI_TEMPLATE(class V, class E)
(requires SemiMovable<V> && SemiMovable<E>)
PUSHMI_TEMPLATE(class... VN, class E)
(requires And<SemiMovable<VN>...> && SemiMovable<E>)
auto error(E e) {
return make_single_sender(detail::single_error_impl<V, E>{std::move(e)});
return make_single_sender(detail::single_error_sender_base{}, detail::single_error_impl<E, VN...>{std::move(e)});
}
} // namespace operators
......
......@@ -9,15 +9,12 @@
#include <tuple>
#include "../piping.h"
#include "../boosters.h"
#include "../single.h"
#include "../sender.h"
#include "../receiver.h"
#include "../flow_receiver.h"
#include "../single_sender.h"
#include "../many.h"
#include "../many_sender.h"
#include "../time_single_sender.h"
#include "../flow_single.h"
#include "../flow_single_sender.h"
#include "../flow_many.h"
#include "../flow_many_sender.h"
#include "../detail/if_constexpr.h"
#include "../detail/functional.h"
......@@ -53,15 +50,13 @@ namespace detail {
template <class Cardinality, bool IsFlow = false>
struct make_receiver;
template <>
struct make_receiver<is_none<>> : construct_deduced<none> {};
struct make_receiver<is_single<>> : construct_deduced<receiver> {};
template <>
struct make_receiver<is_single<>> : construct_deduced<single> {};
struct make_receiver<is_many<>> : construct_deduced<receiver> {};
template <>
struct make_receiver<is_many<>> : construct_deduced<many> {};
struct make_receiver<is_single<>, true> : construct_deduced<flow_receiver> {};
template <>
struct make_receiver<is_single<>, true> : construct_deduced<flow_single> {};
template <>
struct make_receiver<is_many<>, true> : construct_deduced<flow_many> {};
struct make_receiver<is_many<>, true> : construct_deduced<flow_receiver> {};
template <class Cardinality, bool IsFlow>
struct receiver_from_impl {
......@@ -74,13 +69,13 @@ struct receiver_from_impl {
PUSHMI_TEMPLATE (class... Ts, class... Fns,
class This = std::enable_if_t<sizeof...(Fns) != 0, receiver_from_impl>)
(requires And<SemiMovable<Fns>...> &&
Invocable<MakeReceiver, std::tuple<Ts...>> &&
Invocable<This, pushmi::invoke_result_t<MakeReceiver, std::tuple<Ts...>>, Fns...>)
Invocable<MakeReceiver, Ts...> &&
Invocable<This, pushmi::invoke_result_t<MakeReceiver, Ts...>, Fns...>)
auto operator()(std::tuple<Ts...> args, Fns...fns) const {
return This()(This()(std::move(args)), std::move(fns)...);
}
PUSHMI_TEMPLATE(class Out, class...Fns)
(requires Receiver<Out, Cardinality> && And<SemiMovable<Fns>...>)
(requires Receiver<Out> && And<SemiMovable<Fns>...>)
auto operator()(Out out, Fns... fns) const {
return MakeReceiver()(std::move(out), std::move(fns)...);
}
......@@ -89,14 +84,14 @@ struct receiver_from_impl {
template <PUSHMI_TYPE_CONSTRAINT(Sender) In>
using receiver_from_fn =
receiver_from_impl<
property_set_index_t<properties_t<In>, is_silent<>>,
property_set_index_t<properties_t<In>, is_single<>>,
property_query_v<properties_t<In>, is_flow<>>>;
template <class In, class FN>
struct submit_transform_out_1 {
FN fn_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
(requires Receiver<Out> && Invocable<FN, Out> && SenderTo<In, pushmi::invoke_result_t<const FN&, Out>>)
void operator()(In& in, Out out) const {
::pushmi::submit(in, fn_(std::move(out)));
}
......@@ -104,10 +99,10 @@ struct submit_transform_out_1 {
template <class In, class FN>
struct submit_transform_out_2 {
FN fn_;
PUSHMI_TEMPLATE(class TP, class Out)
(requires Receiver<Out>)
void operator()(In& in, TP tp, Out out) const {
::pushmi::submit(in, tp, fn_(std::move(out)));
PUSHMI_TEMPLATE(class CV, class Out)
(requires Receiver<Out> && Invocable<FN, Out> && ConstrainedSenderTo<In, pushmi::invoke_result_t<const FN&, Out>>)
void operator()(In& in, CV cv, Out out) const {
::pushmi::submit(in, cv, fn_(std::move(out)));
}
};
template <class In, class SDSF>
......@@ -122,44 +117,42 @@ struct submit_transform_out_3 {
template <class In, class TSDSF>
struct submit_transform_out_4 {
TSDSF tsdsf_;
PUSHMI_TEMPLATE(class TP, class Out)
(requires Receiver<Out> && Invocable<const TSDSF&, In&, TP, Out>)
void operator()(In& in, TP tp, Out out) const {
tsdsf_(in, tp, std::move(out));
PUSHMI_TEMPLATE(class CV, class Out)
(requires Receiver<Out> && Invocable<const TSDSF&, In&, CV, Out>)
void operator()(In& in, CV cv, Out out) const {
tsdsf_(in, cv, std::move(out));
}
};
PUSHMI_TEMPLATE(class In, class FN)
(requires Sender<In> && SemiMovable<FN>
PUSHMI_BROKEN_SUBSUMPTION(&& not TimeSender<In>))
PUSHMI_BROKEN_SUBSUMPTION(&& not ConstrainedSender<In>))
auto submit_transform_out(FN fn) {
return on_submit(submit_transform_out_1<In, FN>{std::move(fn)});
}
PUSHMI_TEMPLATE(class In, class FN)
(requires TimeSender<In> && SemiMovable<FN>)
(requires ConstrainedSender<In> && SemiMovable<FN>)
auto submit_transform_out(FN fn){
return on_submit(submit_transform_out_2<In, FN>{std::move(fn)});
return submit_transform_out_2<In, FN>{std::move(fn)};
}
PUSHMI_TEMPLATE(class In, class SDSF, class TSDSF)
(requires Sender<In> && SemiMovable<SDSF> && SemiMovable<TSDSF>
PUSHMI_BROKEN_SUBSUMPTION(&& not TimeSender<In>))
PUSHMI_BROKEN_SUBSUMPTION(&& not ConstrainedSender<In>))
auto submit_transform_out(SDSF sdsf, TSDSF) {
return on_submit(submit_transform_out_3<In, SDSF>{std::move(sdsf)});
return submit_transform_out_3<In, SDSF>{std::move(sdsf)};
}
PUSHMI_TEMPLATE(class In, class SDSF, class TSDSF)
(requires TimeSender<In> && SemiMovable<SDSF> && SemiMovable<TSDSF>)
(requires ConstrainedSender<In> && SemiMovable<SDSF> && SemiMovable<TSDSF>)
auto submit_transform_out(SDSF, TSDSF tsdsf) {
return on_submit(submit_transform_out_4<In, TSDSF>{std::move(tsdsf)});
return submit_transform_out_4<In, TSDSF>{std::move(tsdsf)};
}
template <class Cardinality, bool IsConstrained = false, bool IsTime = false, bool IsFlow = false>
struct make_sender;
template <>
struct make_sender<is_none<>> : construct_deduced<sender> {};
template <>
struct make_sender<is_single<>> : construct_deduced<single_sender> {};
template <>
struct make_sender<is_many<>> : construct_deduced<many_sender> {};
......@@ -178,7 +171,7 @@ PUSHMI_INLINE_VAR constexpr struct sender_from_fn {
auto operator()(In in, FN&&... fn) const {
using MakeSender =
make_sender<
property_set_index_t<properties_t<In>, is_silent<>>,
property_set_index_t<properties_t<In>, is_single<>>,
property_query_v<properties_t<In>, is_constrained<>>,
property_query_v<properties_t<In>, is_time<>>,
property_query_v<properties_t<In>, is_flow<>>>;
......@@ -194,10 +187,10 @@ PUSHMI_TEMPLATE(
bool TimeSingleSenderRequires)
(requires Sender<In> && Receiver<Out>)
constexpr bool sender_requires_from() {
PUSHMI_IF_CONSTEXPR_RETURN( ((bool) TimeSenderTo<In, Out, is_single<>>) (
PUSHMI_IF_CONSTEXPR_RETURN( ((bool) TimeSenderTo<In, Out>) (
return TimeSingleSenderRequires;
) else (
PUSHMI_IF_CONSTEXPR_RETURN( ((bool) SenderTo<In, Out, is_single<>>) (
PUSHMI_IF_CONSTEXPR_RETURN( ((bool) SenderTo<In, Out>) (
return SingleSenderRequires;
) else (
PUSHMI_IF_CONSTEXPR_RETURN( ((bool) SenderTo<In, Out>) (
......@@ -210,19 +203,19 @@ constexpr bool sender_requires_from() {
struct set_value_fn {
private:
template <class V>
template <class... VN>
struct impl {
V v_;
std::tuple<VN...> vn_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_single<>>)
(requires ReceiveValue<Out, VN...>)
void operator()(Out out) {
::pushmi::set_value(out, std::move(v_));
::pushmi::apply(::pushmi::set_value, std::tuple_cat(std::tuple<Out>{std::move(out)}, std::move(vn_)));
}
};
public:
template <class V>
auto operator()(V&& v) const {
return impl<std::decay_t<V>>{(V&&) v};
template <class... VN>
auto operator()(VN&&... vn) const {
return impl<std::decay_t<VN>...>{(VN&&) vn...};
}
};
......@@ -232,7 +225,7 @@ private:
struct impl {
E e_;
PUSHMI_TEMPLATE(class Out)
(requires NoneReceiver<Out, E>)
(requires ReceiveError<Out, E>)
void operator()(Out out) {
::pushmi::set_error(out, std::move(e_));
}
......@@ -260,24 +253,6 @@ public:
}
};
struct set_next_fn {
private:
template <class V>
struct impl {
V v_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out, is_many<>>)
void operator()(Out out) {
::pushmi::set_next(out, std::move(v_));
}
};
public:
template <class V>
auto operator()(V&& v) const {
return impl<std::decay_t<V>>{(V&&) v};
}
};
struct set_starting_fn {
private:
template <class Up>
......@@ -383,7 +358,6 @@ namespace extension_operators {
PUSHMI_INLINE_VAR constexpr detail::set_done_fn set_done{};
PUSHMI_INLINE_VAR constexpr detail::set_error_fn set_error{};
PUSHMI_INLINE_VAR constexpr detail::set_value_fn set_value{};
PUSHMI_INLINE_VAR constexpr detail::set_next_fn set_next{};
PUSHMI_INLINE_VAR constexpr detail::set_starting_fn set_starting{};
PUSHMI_INLINE_VAR constexpr detail::executor_fn executor{};
PUSHMI_INLINE_VAR constexpr detail::do_submit_fn submit{};
......
......@@ -13,20 +13,19 @@ namespace detail {
struct filter_fn {
private:
template <class Predicate>
template <class In, class Predicate>
struct on_value_impl {
Predicate p_;
template <class Out, class V>
void operator()(Out& out, V&& v) const {
if (p_(as_const(v))) {
::pushmi::set_value(out, (V&&) v);
} else {
::pushmi::set_done(out);
PUSHMI_TEMPLATE(class Out, class... VN)
(requires Receiver<Out>)
void operator()(Out& out, VN&&... vn) const {
if (p_(as_const(vn)...)) {
::pushmi::set_value(out, (VN&&) vn...);
}
}
};
template <class In, class Predicate>
struct out_impl {
struct submit_impl {
Predicate p_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
......@@ -34,19 +33,19 @@ private:
return ::pushmi::detail::receiver_from_fn<In>()(
std::move(out),
// copy 'p' to allow multiple calls to submit
::pushmi::on_value(on_value_impl<Predicate>{p_})
on_value_impl<In, Predicate>{p_}
);
}
};
template <class Predicate>
struct in_impl {
struct adapt_impl {
Predicate p_;
PUSHMI_TEMPLATE(class In)
(requires Sender<In>)
auto operator()(In in) const {
return ::pushmi::detail::sender_from(
std::move(in),
::pushmi::detail::submit_transform_out<In>(out_impl<In, Predicate>{p_})
::pushmi::detail::submit_transform_out<In>(submit_impl<In, Predicate>{p_})
);
}
};
......@@ -54,7 +53,7 @@ public:
PUSHMI_TEMPLATE(class Predicate)
(requires SemiMovable<Predicate>)
auto operator()(Predicate p) const {
return in_impl<Predicate>{std::move(p)};
return adapt_impl<Predicate>{std::move(p)};
}
};
......
......@@ -20,20 +20,20 @@ private:
using properties = property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>;
std::function<void(ptrdiff_t)> pull;
template<class V>
void next(V&& v) {
::pushmi::set_next(static_cast<Out&>(*this), (V&&) v);
void value(V&& v) {
::pushmi::set_value(static_cast<Out&>(*this), (V&&) v);
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires ManyReceiver<Up, std::ptrdiff_t>)
(requires Receiver<Up>)
void starting(Up up){
pull = [up = std::move(up)](std::ptrdiff_t requested) mutable {
::pushmi::set_next(up, requested);
::pushmi::set_value(up, requested);
};
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires None<Up> && not Many<Up>)
(requires ReceiveValue<Up>)
void starting(Up up){}
};
template <class... AN>
......@@ -42,17 +42,17 @@ private:
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Flow<In> && Many<In>)
In operator()(In in) {
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_silent<>>>>()(std::move(args_))};
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_single<>>>>()(std::move(args_))};
using Out = decltype(out);
::pushmi::submit(in, ::pushmi::detail::receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
return in;
}
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Time<In> && Flow<In> && Many<In>)
(requires Sender<In> && Constrained<In> && Flow<In> && Many<In>)
In operator()(In in) {
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_silent<>>>>()(std::move(args_))};
auto out{::pushmi::detail::receiver_from_fn<subset<is_sender<>, property_set_index_t<properties_t<In>, is_single<>>>>()(std::move(args_))};
using Out = decltype(out);
::pushmi::submit(in, ::pushmi::now(in), ::pushmi::detail::receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
::pushmi::submit(in, ::pushmi::top(in), ::pushmi::detail::receiver_from_fn<In>()(Pull<In, Out>{std::move(out)}));
return in;
}
};
......
......@@ -24,16 +24,19 @@ namespace operators {
PUSHMI_INLINE_VAR constexpr struct from_fn {
private:
struct sender_base : many_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_many<>, is_always_blocking<>, is_fifo_sequence<>>;
};
template <class I, class S>
struct out_impl {
I begin_;
S end_;
PUSHMI_TEMPLATE(class Out)
(requires ManyReceiver<Out, typename std::iterator_traits<I>::value_type>)
void operator()(Out out) const {
(requires ReceiveValue<Out, typename std::iterator_traits<I>::value_type>)
void operator()(sender_base&, Out out) const {
auto c = begin_;
for (; c != end_; ++c) {
::pushmi::set_next(out, *c);
::pushmi::set_value(out, *c);
}
::pushmi::set_done(out);
}
......@@ -45,7 +48,7 @@ public:
typename std::iterator_traits<I>::iterator_category,
std::forward_iterator_tag>)
auto operator()(I begin, S end) const {
return make_many_sender(out_impl<I, S>{begin, end});
return make_many_sender(sender_base{}, out_impl<I, S>{begin, end});
}
PUSHMI_TEMPLATE(class R)
......@@ -72,24 +75,24 @@ struct flow_from_producer {
template<class Producer>
struct flow_from_up {
using properties = properties_t<many<>>;
using properties = properties_t<receiver<>>;
explicit flow_from_up(std::shared_ptr<Producer> p) : p(std::move(p)) {}
std::shared_ptr<Producer> p;
void next(ptrdiff_t requested) {
void value(ptrdiff_t requested) {
if (requested < 1) {return;}
// submit work to exec
::pushmi::submit(p->exec,
make_single([p = p, requested](auto) {
make_receiver([p = p, requested](auto) {
auto remaining = requested;
// this loop is structured to work when there is re-entrancy
// out.next in the loop may call up.next. to handle this the
// out.value in the loop may call up.value. to handle this the
// state of p->c must be captured and the remaining and p->c
// must be changed before out.next is called.
// must be changed before out.value is called.
while (remaining-- > 0 && !p->stop && p->c != p->end) {
auto i = (p->c)++;
::pushmi::set_next(p->out, ::pushmi::detail::as_const(*i));
::pushmi::set_value(p->out, ::pushmi::detail::as_const(*i));
}
if (p->c == p->end) {
::pushmi::set_done(p->out);
......@@ -101,7 +104,7 @@ struct flow_from_up {
void error(E e) noexcept {
p->stop.store(true);
::pushmi::submit(p->exec,
make_single([p = p](auto) {
make_receiver([p = p](auto) {
::pushmi::set_done(p->out);
}));
}
......@@ -109,7 +112,7 @@ struct flow_from_up {
void done() {
p->stop.store(true);
::pushmi::submit(p->exec,
make_single([p = p](auto) {
make_receiver([p = p](auto) {
::pushmi::set_done(p->out);
}));
}
......@@ -123,15 +126,15 @@ private:
S end_;
mutable Exec exec_;
PUSHMI_TEMPLATE(class Out)
(requires ManyReceiver<Out, typename std::iterator_traits<I>::value_type>)
(requires ReceiveValue<Out, typename std::iterator_traits<I>::value_type>)
void operator()(Out out) const {
using Producer = flow_from_producer<I, S, Out, Exec>;
auto p = std::make_shared<Producer>(begin_, end_, std::move(out), exec_, false);
::pushmi::submit(exec_,
make_single([p](auto exec) {
make_receiver([p](auto exec) {
// pass reference for cancellation.
::pushmi::set_starting(p->out, make_many(flow_from_up<Producer>{p}));
::pushmi::set_starting(p->out, make_receiver(flow_from_up<Producer>{p}));
}));
}
};
......
......@@ -16,20 +16,24 @@ namespace operators {
PUSHMI_INLINE_VAR constexpr struct just_fn {
private:
template <class V>
struct sender_base : single_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_single<>, is_always_blocking<>, is_fifo_sequence<>>;
};
template <class... VN>
struct impl {
V v_;
std::tuple<VN...> vn_;
PUSHMI_TEMPLATE (class Out)
(requires SingleReceiver<Out, V>)
void operator()(Out out) {
::pushmi::set_value(out, std::move(v_));
(requires ReceiveValue<Out, VN...>)
void operator()(sender_base&, Out out) {
::pushmi::apply(::pushmi::set_value, std::tuple_cat(std::tuple<Out&>{out}, std::move(vn_)));
::pushmi::set_done(std::move(out));
}
};
public:
PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
auto operator()(V v) const {
return make_single_sender(impl<V>{std::move(v)});
PUSHMI_TEMPLATE(class... VN)
(requires And<SemiMovable<VN>...>)
auto operator()(VN... vn) const {
return make_single_sender(sender_base{}, impl<VN...>{std::tuple<VN...>{std::move(vn)...}});
}
} just {};
} // namespace operators
......
......@@ -6,7 +6,7 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "../single.h"
#include "../receiver.h"
#include "submit.h"
#include "extension_operators.h"
#include "via.h"
......@@ -16,7 +16,9 @@ namespace pushmi {
template<typename In>
struct send_via {
In in;
template<class... AN>
PUSHMI_TEMPLATE(class... AN)
(requires Invocable<decltype(::pushmi::operators::via), AN...> &&
Invocable<invoke_result_t<decltype(::pushmi::operators::via), AN...>, In>)
auto via(AN&&... an) {
return in | ::pushmi::operators::via((AN&&) an...);
}
......
......@@ -6,7 +6,7 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "../single.h"
#include "../receiver.h"
#include "submit.h"
#include "extension_operators.h"
......@@ -16,19 +16,15 @@ namespace pushmi {
namespace detail {
template<class T>
template<class... TN>
struct share_fn {
private:
struct impl {
PUSHMI_TEMPLATE (class In)
(requires Sender<In>)
auto operator()(In in) const {
subject<T, properties_t<In>> sub;
PUSHMI_IF_CONSTEXPR( ((bool)TimeSender<In>) (
::pushmi::submit(in, ::pushmi::now(id(in)), sub.receiver());
) else (
::pushmi::submit(id(in), sub.receiver());
));
subject<properties_t<In>, TN...> sub;
::pushmi::submit(in, sub.receiver());
return sub;
}
};
......@@ -42,8 +38,8 @@ public:
namespace operators {
template<class T>
PUSHMI_INLINE_VAR constexpr detail::share_fn<T> share{};
template<class... TN>
PUSHMI_INLINE_VAR constexpr detail::share_fn<TN...> share{};
} // namespace operators
......
......@@ -16,7 +16,8 @@ private:
template <class ErrorSelector>
struct on_error_impl {
ErrorSelector es_;
template <class Out, class E>
PUSHMI_TEMPLATE (class Out, class E)
(requires Receiver<Out> && Invocable<const ErrorSelector&, E> && SenderTo<pushmi::invoke_result_t<ErrorSelector&, E>, Out>)
void operator()(Out& out, E&& e) const noexcept {
static_assert(::pushmi::NothrowInvocable<const ErrorSelector&, E>,
"switch_on_error - error selector function must be noexcept");
......
......@@ -172,4 +172,30 @@ template<class PS, class... ExpectedN>
PUSHMI_INLINE_VAR constexpr bool property_query_v =
property_query<PS, ExpectedN...>::value;
// query for categories on types with properties.
namespace detail {
template<class CIn, class POut>
std::true_type category_query_fn(property_set_element<POut, CIn>*);
template<class C>
std::false_type category_query_fn(void*);
template<class PS, class... ExpectedN>
struct category_query_impl : bool_<
and_v<decltype(category_query_fn<ExpectedN>(
(properties_t<PS>*)nullptr))::value...>> {};
} //namespace detail
template<class PS, class... ExpectedN>
struct category_query
: std::conditional_t<
Properties<PS> && not Or<Property<ExpectedN>...>,
detail::category_query_impl<PS, ExpectedN...>,
std::false_type> {};
template<class PS, class... ExpectedN>
PUSHMI_INLINE_VAR constexpr bool category_query_v =
category_query<PS, ExpectedN...>::value;
} // namespace pushmi
This diff is collapsed.
This diff is collapsed.
......@@ -4,7 +4,7 @@
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
#include "single.h"
#include "receiver.h"
#include "executor.h"
#include "inline.h"
#include "constrained_single_sender.h"
......@@ -12,20 +12,20 @@
namespace pushmi {
template <class V, class E, class TP>
class any_time_single_sender : public any_constrained_single_sender<V, E, TP> {
template <class E, class TP, class... VN>
class any_time_single_sender : public any_constrained_single_sender<E, TP, VN...> {
public:
using properties = property_set<is_time<>, is_single<>>;
constexpr any_time_single_sender() = default;
template<class T>
constexpr explicit any_time_single_sender(T t)
: any_constrained_single_sender<V, E, TP>(std::move(t)) {}
: any_constrained_single_sender<E, TP, VN...>(std::move(t)) {}
template<class T0, class T1, class... TN>
constexpr any_time_single_sender(T0 t0, T1 t1, TN... tn)
: any_constrained_single_sender<V, E, TP>(std::move(t0), std::move(t1), std::move(tn)...) {}
: any_constrained_single_sender<E, TP, VN...>(std::move(t0), std::move(t1), std::move(tn)...) {}
any_time_executor<E, TP> executor() {
return any_time_executor<E, TP>{any_constrained_single_sender<V, E, TP>::executor()};
return any_time_executor<E, TP>{any_constrained_single_sender<E, TP, VN...>::executor()};
}
};
......
This diff is collapsed.
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