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

add new operators to single header (#18)

fbshipit-source-id: 1df9cc44331327b62fb7e20f22f8f512fe4e6ecd
parent 3e05ecc2
...@@ -45,14 +45,18 @@ set(header_files ...@@ -45,14 +45,18 @@ set(header_files
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_deferred.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/flow_single_deferred.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/trampoline.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/new_thread.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/extension_operators.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/defer.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/on.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/on.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/submit.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/tap.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/tap.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/transform.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/transform.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/via.h" "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/via.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/request_via.h"
"${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/share.h"
) )
BuildSingleHeader("pushmi" ${header_files}) BuildSingleHeader("pushmi" ${header_files})
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <future> #include <future>
#include <tuple> #include <tuple>
#include <deque> #include <deque>
#include <vector>
#if __cpp_lib_optional >= 201606 #if __cpp_lib_optional >= 201606
#include <optional> #include <optional>
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <future> #include <future>
#include <tuple> #include <tuple>
#include <deque> #include <deque>
#include <vector>
#if __cpp_lib_optional >= 201606 #if __cpp_lib_optional >= 201606
#include <optional> #include <optional>
...@@ -8311,43 +8312,6 @@ inline auto new_thread() { ...@@ -8311,43 +8312,6 @@ inline auto new_thread() {
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
//#include <meta/meta.hpp>
//#include "../deferred.h"
//#include "../single_deferred.h"
//#include "../detail/functional.h"
namespace pushmi {
namespace operators {
template <class V>
auto empty() {
return make_single_deferred(
constrain(lazy::SingleReceiver<_1, V>, [](auto out) mutable {
::pushmi::set_done(out);
})
);
}
inline auto empty() {
return make_deferred(
constrain(lazy::NoneReceiver<_1>, [](auto out) mutable {
::pushmi::set_done(out);
})
);
}
} // namespace operators
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include <tuple> //#include <tuple>
//#include "../piping.h" //#include "../piping.h"
//#include "../boosters.h" //#include "../boosters.h"
...@@ -8624,228 +8588,358 @@ PUSHMI_INLINE_VAR constexpr detail::now_fn top{}; ...@@ -8624,228 +8588,358 @@ PUSHMI_INLINE_VAR constexpr detail::now_fn top{};
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
//#include "../single_deferred.h" //#include <functional>
//#include "submit.h" //#include "../time_single_deferred.h"
//#include "extension_operators.h" //#include "../boosters.h"
namespace pushmi {
namespace operators {
PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
auto just(V v) {
return make_single_deferred(
constrain(lazy::SingleReceiver<_1, V>,
[v = std::move(v)](auto out) mutable {
::pushmi::set_value(out, std::move(v));
}
)
);
}
} // namespace operators
} // namespace pushmi
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include "../piping.h"
//#include "../executor.h"
//#include "extension_operators.h" //#include "extension_operators.h"
//#include "../trampoline.h"
//#include "../detail/opt.h"
//#include "../detail/if_constexpr.h"
namespace pushmi { namespace pushmi {
namespace detail { namespace detail {
namespace submit_detail {
template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN>
using receiver_type_t =
pushmi::invoke_result_t<
pushmi::detail::make_receiver<property_set_index_t<properties_t<In>, is_silent<>>>,
AN...>;
struct on_fn { PUSHMI_CONCEPT_DEF(
PUSHMI_TEMPLATE(class ExecutorFactory) template (class In, class ... AN)
(requires Invocable<ExecutorFactory&>) (concept AutoSenderTo)(In, AN...),
auto operator()(ExecutorFactory ef) const { SenderTo<In, receiver_type_t<In, AN...>>
return constrain(lazy::Sender<_1>, [ef = std::move(ef)](auto in) { );
using In = decltype(in); PUSHMI_CONCEPT_DEF(
return ::pushmi::detail::deferred_from<In, single<>>( template (class In, class ... AN)
std::move(in), (concept AutoTimeSenderTo)(In, AN...),
::pushmi::detail::submit_transform_out<In>( TimeSenderTo<In, receiver_type_t<In, AN...>>
constrain(lazy::SenderTo<In, _2>, [ef](In& in, auto out) { );
auto exec = ef(); } // namespace submit_detail
::pushmi::submit(exec, ::pushmi::now(exec),
::pushmi::make_single([in = in, out = std::move(out)](auto) mutable { struct submit_fn {
private:
// TODO - only move, move-only types..
// if out can be copied, then submit can be called multiple
// times..
template <class... AN>
struct fn {
std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoSenderTo<In, AN...>)
In operator()(In in) {
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
::pushmi::submit(in, std::move(out)); ::pushmi::submit(in, std::move(out));
}) return in;
); }
}), PUSHMI_TEMPLATE(class In)
constrain(lazy::TimeSenderTo<In, _3>, [ef](In& in, auto at, auto out) { (requires submit_detail::AutoTimeSenderTo<In, AN...>)
auto exec = ef(); In operator()(In in) {
::pushmi::submit(exec, at, auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
::pushmi::on_value([in = in, at, out = std::move(out)](auto) mutable { ::pushmi::submit(in, ::pushmi::now(in), std::move(out));
::pushmi::submit(in, at, std::move(out)); return in;
}) }
); };
}) public:
) template <class... AN>
); auto operator()(AN&&... an) const {
}); return submit_fn::fn<AN...>{{(AN&&) an...}};
} }
}; };
} // namespace detail struct submit_at_fn {
private:
namespace operators { template <class TP, class...AN>
struct fn {
PUSHMI_INLINE_VAR constexpr detail::on_fn on{}; TP at_;
std::tuple<AN...> args_;
} // namespace operators PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoTimeSenderTo<In, AN...>)
#if 0 In operator()(In in) {
namespace detail { auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
::pushmi::submit(in, std::move(at_), std::move(out));
template <class ExecutorFactory> return in;
class fsdon {
using executor_factory_type = std::decay_t<ExecutorFactory>;
executor_factory_type factory_;
template <class In>
class start_on {
using in_type = std::decay_t<In>;
executor_factory_type factory_;
in_type in_;
template <class Out, class Executor>
class out_on {
using out_type = std::decay_t<Out>;
using exec_type = std::decay_t<Executor>;
template <class Producer>
struct producer_proxy {
RefWrapper<Producer> up_;
std::shared_ptr<std::atomic_bool> stopped_;
exec_type exec_;
producer_proxy(
RefWrapper<Producer> p,
std::shared_ptr<std::atomic_bool> stopped,
exec_type exec)
: up_(std::move(p)),
stopped_(std::move(stopped)),
exec_(std::move(exec)) {}
template <class V>
void value(V v) {
auto up = wrap_ref(up_.get());
exec_ |
execute([up = std::move(up),
v = std::move(v),
stopped = std::move(stopped_)](auto) mutable {
if (*stopped) {
return;
} }
up.get().value(std::move(v)); };
}); public:
PUSHMI_TEMPLATE(class TP, class...AN)
(requires Regular<TP>)
auto operator()(TP at, AN... an) const {
return submit_at_fn::fn<TP, AN...>{std::move(at), {(AN&&) an...}};
} }
};
template <class E> struct submit_after_fn {
void error(E e) { private:
auto up = wrap_ref(up_.get()); template <class D, class... AN>
exec_ | struct fn {
execute([up = std::move(up), D after_;
e = std::move(e), std::tuple<AN...> args_;
stopped = std::move(stopped_)](auto) mutable { PUSHMI_TEMPLATE(class In)
if (*stopped) { (requires submit_detail::AutoTimeSenderTo<In, AN...>)
return; In operator()(In in) {
} // TODO - only move, move-only types..
up.get().error(std::move(e)); // if out can be copied, then submit can be called multiple
}); // times..
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
auto at = ::pushmi::now(in) + std::move(after_);
::pushmi::submit(in, std::move(at), std::move(out));
return in;
} }
}; };
public:
PUSHMI_TEMPLATE(class D, class...AN)
(requires Regular<D>)
auto operator()(D after, AN... an) const {
return submit_after_fn::fn<D, AN...>{std::move(after), {(AN&&) an...}};
}
};
bool done_; struct blocking_submit_fn {
std::shared_ptr<std::atomic_bool> stopped_; private:
out_type out_; // TODO - only move, move-only types..
exec_type exec_; // if out can be copied, then submit can be called multiple
AnyNone<> upProxy_; // times..
template <class... AN>
public: struct fn {
out_on(out_type out, exec_type exec) std::tuple<AN...> args_;
: done_(false),
stopped_(std::make_shared<std::atomic_bool>(false)),
out_(std::move(out)),
exec_(std::move(exec)),
upProxy_() {}
template <class T> template <bool IsTimeSender, class In>
void value(T t) { In impl_(In in) {
if (done_) { bool done = false;
return; std::condition_variable signaled;
auto out{::pushmi::detail::out_from_fn<In>()(
std::move(args_),
on_value(constrain(pushmi::lazy::Receiver<_1, is_single<>>,
[&](auto out, auto&& v) {
using V = remove_cvref_t<decltype(v)>;
PUSHMI_IF_CONSTEXPR( ((bool)Time<V>) (
// to keep the blocking semantics, make sure that the
// nested submits block here to prevent a spurious
// completion signal
auto nest = ::pushmi::nested_trampoline();
::pushmi::submit(nest, ::pushmi::now(nest), std::move(out));
) else (
::pushmi::set_value(out, id((V&&) v));
))
done = true;
signaled.notify_all();
} }
done_ = true; )),
out_.value(std::move(t)); on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>,
[&](auto out, auto e) noexcept {
::pushmi::set_error(out, std::move(e));
done = true;
signaled.notify_all();
} }
)),
template <class E> on_done(constrain(pushmi::lazy::Receiver<_1>,
void error(E e) { [&](auto out){
if (done_) { ::pushmi::set_done(out);
return; done = true;
signaled.notify_all();
} }
done_ = true; ))
out_.error(std::move(e)); )};
PUSHMI_IF_CONSTEXPR( (IsTimeSender) (
id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out));
) else (
id(::pushmi::submit)(in, std::move(out));
))
std::mutex lock;
std::unique_lock<std::mutex> guard{lock};
signaled.wait(guard, [&]{
return done;
});
return in;
} }
void stopping() { PUSHMI_TEMPLATE(class In)
if (done_) { (requires submit_detail::AutoSenderTo<In, AN...>)
return; In operator()(In in) {
return this->impl_<false>(std::move(in));
} }
done_ = true; PUSHMI_TEMPLATE(class In)
*stopped_ = true; (requires submit_detail::AutoTimeSenderTo<In, AN...>)
out_.stopping(); In operator()(In in) {
return this->impl_<true>(std::move(in));
}
};
public:
template <class... AN>
auto operator()(AN... an) const {
return blocking_submit_fn::fn<AN...>{{(AN&&) an...}};
} }
};
template <class Producer> template <class T>
void starting(RefWrapper<Producer> up) { struct get_fn {
upProxy_ = // TODO constrain this better
producer_proxy<Producer>{std::move(up), stopped_, std::move(exec_)}; PUSHMI_TEMPLATE (class In)
out_.starting(wrap_ref(upProxy_)); (requires Sender<In>)
T operator()(In in) const {
pushmi::detail::opt<T> result_;
std::exception_ptr ep_;
auto out = make_single(
on_value([&](T t){ result_ = std::move(t); }),
on_error(
[&](auto e) noexcept { ep_ = std::make_exception_ptr(e); },
[&](std::exception_ptr ep) noexcept { ep_ = ep; })
);
using Out = decltype(out);
static_assert(SenderTo<In, Out, is_single<>> ||
TimeSenderTo<In, Out, is_single<>>,
"'In' does not deliver value compatible with 'T' to 'Out'");
blocking_submit_fn{}(std::move(out))(in);
if (!!ep_) { std::rethrow_exception(ep_); }
return std::move(*result_);
}
};
} // namespace detail
namespace operators {
PUSHMI_INLINE_VAR constexpr detail::submit_fn submit{};
PUSHMI_INLINE_VAR constexpr detail::submit_at_fn submit_at{};
PUSHMI_INLINE_VAR constexpr detail::submit_after_fn submit_after{};
PUSHMI_INLINE_VAR constexpr detail::blocking_submit_fn blocking_submit{};
template <class T>
PUSHMI_INLINE_VAR constexpr detail::get_fn<T> get{};
} // namespace operators
} // namespace pushmi
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include <vector>
//#include <pushmi/time_single_deferred.h>
namespace pushmi {
template<class... TN>
struct subject;
template<class T, class PS>
struct subject<T, PS> {
using properties = property_set_insert_t<property_set<is_sender<>, is_single<>>, PS>;
struct subject_shared {
bool done_ = false;
pushmi::detail::opt<T> t_;
std::exception_ptr ep_;
std::vector<any_single<T>> receivers_;
std::mutex lock_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
void submit(Out out) {
std::unique_lock<std::mutex> guard(lock_);
if (ep_) {::pushmi::set_error(out, ep_); return;}
if (!!t_) {::pushmi::set_value(out, (const T&)t_); return;}
if (done_) {::pushmi::set_done(out); return;}
receivers_.push_back(any_single<T>{out});
}
PUSHMI_TEMPLATE(class V)
(requires SemiMovable<V>)
void value(V&& v) {
std::unique_lock<std::mutex> guard(lock_);
t_ = detail::as_const(v);
for (auto& out : receivers_) {::pushmi::set_value(out, (V&&) v);}
receivers_.clear();
}
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>)
void error(E e) noexcept {
std::unique_lock<std::mutex> guard(lock_);
ep_ = e;
for (auto& out : receivers_) {::pushmi::set_error(out, std::move(e));}
receivers_.clear();
}
void done() {
std::unique_lock<std::mutex> guard(lock_);
done_ = true;
for (auto& out : receivers_) {::pushmi::set_done(out);}
receivers_.clear();
} }
}; };
public: // need a template overload of none/deferred and the rest that stores a 'ptr' with its own lifetime management
start_on(executor_factory_type&& ef, in_type&& in) struct subject_receiver {
: factory_(std::move(ef)), in_(std::move(in)) {}
template <class Out> using properties = property_set_insert_t<property_set<is_receiver<>, is_single<>>, PS>;
auto then(Out out) {
auto exec = factory_(); std::shared_ptr<subject_shared> s;
auto myout = out_on<Out, decltype(exec)>{std::move(out), exec};
exec | execute([in = in_, myout = std::move(myout)](auto) mutable { PUSHMI_TEMPLATE(class V)
in.then(std::move(myout)); (requires SemiMovable<V>)
}); void value(V&& v) {
s->value((V&&) v);
}
PUSHMI_TEMPLATE(class E)
(requires SemiMovable<E>)
void error(E e) noexcept {
s->error(std::move(e));
}
void done() {
s->done();
} }
}; };
public: std::shared_ptr<subject_shared> s = std::make_shared<subject_shared>();
explicit fsdon(executor_factory_type&& ef) : factory_(std::move(ef)) {}
template <class In> PUSHMI_TEMPLATE(class Out)
auto operator()(In in) { (requires Receiver<Out>)
return start_on<In>{std::move(factory_), std::move(in)}; void submit(Out out) {
s->submit(std::move(out));
}
auto receiver() {
return detail::out_from_fn<subject>{}(subject_receiver{s});
} }
}; };
} // namespace detail } // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
namespace fsd { //#include <meta/meta.hpp>
template <class ExecutorFactory>
auto on(ExecutorFactory factory) { //#include "../deferred.h"
return detail::fsdon<ExecutorFactory>{std::move(factory)}; //#include "../single_deferred.h"
//#include "../detail/functional.h"
namespace pushmi {
namespace operators {
template <class V>
auto empty() {
return make_single_deferred(
constrain(lazy::SingleReceiver<_1, V>, [](auto out) mutable {
::pushmi::set_done(out);
})
);
}
inline auto empty() {
return make_deferred(
constrain(lazy::NoneReceiver<_1>, [](auto out) mutable {
::pushmi::set_done(out);
})
);
} }
} // namespace fsd
#endif
} // namespace operators
} // namespace pushmi } // namespace pushmi
// clang-format off // clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{} // clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
...@@ -8855,224 +8949,270 @@ auto on(ExecutorFactory factory) { ...@@ -8855,224 +8949,270 @@ auto on(ExecutorFactory factory) {
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
//#include <functional> //#include "../single_deferred.h"
//#include "../time_single_deferred.h" //#include "submit.h"
//#include "../boosters.h"
//#include "extension_operators.h" //#include "extension_operators.h"
//#include "../trampoline.h"
//#include "../detail/opt.h"
//#include "../detail/if_constexpr.h"
namespace pushmi { namespace pushmi {
namespace detail {
namespace submit_detail {
template <PUSHMI_TYPE_CONSTRAINT(Sender) In, class ...AN>
using receiver_type_t =
pushmi::invoke_result_t<
pushmi::detail::make_receiver<property_set_index_t<properties_t<In>, is_silent<>>>,
AN...>;
PUSHMI_CONCEPT_DEF( namespace operators {
template (class In, class ... AN)
(concept AutoSenderTo)(In, AN...),
SenderTo<In, receiver_type_t<In, AN...>>
);
PUSHMI_CONCEPT_DEF(
template (class In, class ... AN)
(concept AutoTimeSenderTo)(In, AN...),
TimeSenderTo<In, receiver_type_t<In, AN...>>
);
} // namespace submit_detail
struct submit_fn { PUSHMI_TEMPLATE(class V)
private: (requires SemiMovable<V>)
// TODO - only move, move-only types.. auto just(V v) {
// if out can be copied, then submit can be called multiple return make_single_deferred(
// times.. constrain(lazy::SingleReceiver<_1, V>,
template <class... AN> [v = std::move(v)](auto out) mutable {
struct fn { ::pushmi::set_value(out, std::move(v));
std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoSenderTo<In, AN...>)
In operator()(In in) {
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
::pushmi::submit(in, std::move(out));
return in;
}
PUSHMI_TEMPLATE(class In)
(requires submit_detail::AutoTimeSenderTo<In, AN...>)
In operator()(In in) {
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))};
::pushmi::submit(in, ::pushmi::now(in), std::move(out));
return in;
}
};
public:
template <class... AN>
auto operator()(AN&&... an) const {
return submit_fn::fn<AN...>{{(AN&&) an...}};
} }
}; )
);
}
struct submit_at_fn { } // namespace operators
private:
template <class TP, class...AN> } // namespace pushmi
struct fn { // clang-format off
TP at_; // clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
std::tuple<AN...> args_; //#pragma once
PUSHMI_TEMPLATE(class In) // Copyright (c) 2018-present, Facebook, Inc.
(requires submit_detail::AutoTimeSenderTo<In, AN...>) //
In operator()(In in) { // This source code is licensed under the MIT license found in the
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))}; // LICENSE file in the root directory of this source tree.
::pushmi::submit(in, std::move(at_), std::move(out));
return in; //#include "../single.h"
//#include "submit.h"
//#include "extension_operators.h"
#if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_
#define MAKE_(...) {__VA_ARGS__}
#else
#define MAKE(x) make_ ## x
#endif
namespace pushmi {
namespace operators {
PUSHMI_TEMPLATE(class F)
(requires Invocable<F>)
auto defer(F f) {
return MAKE(single_deferred)(
constrain(lazy::Receiver<_1>,
[f = std::move(f)](auto out) mutable {
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 TP, class...AN) }
(requires Regular<TP>)
auto operator()(TP at, AN... an) const { } // namespace operators
return submit_at_fn::fn<TP, AN...>{std::move(at), {(AN&&) an...}};
} // namespace pushmi
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include "../piping.h"
//#include "../executor.h"
//#include "extension_operators.h"
namespace pushmi {
namespace detail {
struct on_fn {
PUSHMI_TEMPLATE(class ExecutorFactory)
(requires Invocable<ExecutorFactory&>)
auto operator()(ExecutorFactory ef) const {
return constrain(lazy::Sender<_1>, [ef = std::move(ef)](auto in) {
using In = decltype(in);
return ::pushmi::detail::deferred_from<In, single<>>(
std::move(in),
::pushmi::detail::submit_transform_out<In>(
constrain(lazy::SenderTo<In, _2>, [ef](In& in, auto out) {
auto exec = ef();
::pushmi::submit(exec, ::pushmi::now(exec),
::pushmi::make_single([in = in, out = std::move(out)](auto) mutable {
::pushmi::submit(in, std::move(out));
})
);
}),
constrain(lazy::TimeSenderTo<In, _3>, [ef](In& in, auto at, auto out) {
auto exec = ef();
::pushmi::submit(exec, at,
::pushmi::on_value([in = in, at, out = std::move(out)](auto) mutable {
::pushmi::submit(in, at, std::move(out));
})
);
})
)
);
});
} }
}; };
struct submit_after_fn { } // namespace detail
private:
template <class D, class... AN> namespace operators {
struct fn {
D after_; PUSHMI_INLINE_VAR constexpr detail::on_fn on{};
std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In) } // namespace operators
(requires submit_detail::AutoTimeSenderTo<In, AN...>)
In operator()(In in) { #if 0
// TODO - only move, move-only types.. namespace detail {
// if out can be copied, then submit can be called multiple
// times.. template <class ExecutorFactory>
auto out{::pushmi::detail::out_from_fn<In>()(std::move(args_))}; class fsdon {
auto at = ::pushmi::now(in) + std::move(after_); using executor_factory_type = std::decay_t<ExecutorFactory>;
::pushmi::submit(in, std::move(at), std::move(out));
return in; executor_factory_type factory_;
template <class In>
class start_on {
using in_type = std::decay_t<In>;
executor_factory_type factory_;
in_type in_;
template <class Out, class Executor>
class out_on {
using out_type = std::decay_t<Out>;
using exec_type = std::decay_t<Executor>;
template <class Producer>
struct producer_proxy {
RefWrapper<Producer> up_;
std::shared_ptr<std::atomic_bool> stopped_;
exec_type exec_;
producer_proxy(
RefWrapper<Producer> p,
std::shared_ptr<std::atomic_bool> stopped,
exec_type exec)
: up_(std::move(p)),
stopped_(std::move(stopped)),
exec_(std::move(exec)) {}
template <class V>
void value(V v) {
auto up = wrap_ref(up_.get());
exec_ |
execute([up = std::move(up),
v = std::move(v),
stopped = std::move(stopped_)](auto) mutable {
if (*stopped) {
return;
}
up.get().value(std::move(v));
});
} }
};
public: template <class E>
PUSHMI_TEMPLATE(class D, class...AN) void error(E e) {
(requires Regular<D>) auto up = wrap_ref(up_.get());
auto operator()(D after, AN... an) const { exec_ |
return submit_after_fn::fn<D, AN...>{std::move(after), {(AN&&) an...}}; execute([up = std::move(up),
e = std::move(e),
stopped = std::move(stopped_)](auto) mutable {
if (*stopped) {
return;
} }
}; up.get().error(std::move(e));
});
}
};
struct blocking_submit_fn { bool done_;
private: std::shared_ptr<std::atomic_bool> stopped_;
// TODO - only move, move-only types.. out_type out_;
// if out can be copied, then submit can be called multiple exec_type exec_;
// times.. AnyNone<> upProxy_;
template <class... AN>
struct fn {
std::tuple<AN...> args_;
template <bool IsTimeSender, class In> public:
In impl_(In in) { out_on(out_type out, exec_type exec)
bool done = false; : done_(false),
std::condition_variable signaled; stopped_(std::make_shared<std::atomic_bool>(false)),
auto out{::pushmi::detail::out_from_fn<In>()( out_(std::move(out)),
std::move(args_), exec_(std::move(exec)),
on_value(constrain(pushmi::lazy::Receiver<_1, is_single<>>, upProxy_() {}
[&](auto out, auto&& v) {
using V = remove_cvref_t<decltype(v)>; template <class T>
PUSHMI_IF_CONSTEXPR( ((bool)Time<V>) ( void value(T t) {
// to keep the blocking semantics, make sure that the if (done_) {
// nested submits block here to prevent a spurious return;
// completion signal
auto nest = ::pushmi::nested_trampoline();
::pushmi::submit(nest, ::pushmi::now(nest), std::move(out));
) else (
::pushmi::set_value(out, id((V&&) v));
))
done = true;
signaled.notify_all();
} }
)), done_ = true;
on_error(constrain(pushmi::lazy::NoneReceiver<_1, _2>, out_.value(std::move(t));
[&](auto out, auto e) noexcept {
::pushmi::set_error(out, std::move(e));
done = true;
signaled.notify_all();
} }
)),
on_done(constrain(pushmi::lazy::Receiver<_1>, template <class E>
[&](auto out){ void error(E e) {
::pushmi::set_done(out); if (done_) {
done = true; return;
signaled.notify_all();
} }
)) done_ = true;
)}; out_.error(std::move(e));
PUSHMI_IF_CONSTEXPR( (IsTimeSender) (
id(::pushmi::submit)(in, id(::pushmi::now)(in), std::move(out));
) else (
id(::pushmi::submit)(in, std::move(out));
))
std::mutex lock;
std::unique_lock<std::mutex> guard{lock};
signaled.wait(guard, [&]{
return done;
});
return in;
} }
PUSHMI_TEMPLATE(class In) void stopping() {
(requires submit_detail::AutoSenderTo<In, AN...>) if (done_) {
In operator()(In in) { return;
return this->impl_<false>(std::move(in));
} }
PUSHMI_TEMPLATE(class In) done_ = true;
(requires submit_detail::AutoTimeSenderTo<In, AN...>) *stopped_ = true;
In operator()(In in) { out_.stopping();
return this->impl_<true>(std::move(in)); }
template <class Producer>
void starting(RefWrapper<Producer> up) {
upProxy_ =
producer_proxy<Producer>{std::move(up), stopped_, std::move(exec_)};
out_.starting(wrap_ref(upProxy_));
} }
}; };
public:
template <class... AN> public:
auto operator()(AN... an) const { start_on(executor_factory_type&& ef, in_type&& in)
return blocking_submit_fn::fn<AN...>{{(AN&&) an...}}; : factory_(std::move(ef)), in_(std::move(in)) {}
template <class Out>
auto then(Out out) {
auto exec = factory_();
auto myout = out_on<Out, decltype(exec)>{std::move(out), exec};
exec | execute([in = in_, myout = std::move(myout)](auto) mutable {
in.then(std::move(myout));
});
} }
}; };
template <class T> public:
struct get_fn { explicit fsdon(executor_factory_type&& ef) : factory_(std::move(ef)) {}
// TODO constrain this better
PUSHMI_TEMPLATE (class In) template <class In>
(requires Sender<In>) auto operator()(In in) {
T operator()(In in) const { return start_on<In>{std::move(factory_), std::move(in)};
pushmi::detail::opt<T> result_;
std::exception_ptr ep_;
auto out = make_single(
on_value([&](T t){ result_ = std::move(t); }),
on_error(
[&](auto e) noexcept { ep_ = std::make_exception_ptr(e); },
[&](std::exception_ptr ep) noexcept { ep_ = ep; })
);
using Out = decltype(out);
static_assert(SenderTo<In, Out, is_single<>> ||
TimeSenderTo<In, Out, is_single<>>,
"'In' does not deliver value compatible with 'T' to 'Out'");
blocking_submit_fn{}(std::move(out))(in);
if (!!ep_) { std::rethrow_exception(ep_); }
return std::move(*result_);
} }
}; };
} // namespace detail } // namespace detail
namespace operators { namespace fsd {
PUSHMI_INLINE_VAR constexpr detail::submit_fn submit{}; template <class ExecutorFactory>
PUSHMI_INLINE_VAR constexpr detail::submit_at_fn submit_at{}; auto on(ExecutorFactory factory) {
PUSHMI_INLINE_VAR constexpr detail::submit_after_fn submit_after{}; return detail::fsdon<ExecutorFactory>{std::move(factory)};
PUSHMI_INLINE_VAR constexpr detail::blocking_submit_fn blocking_submit{}; }
template <class T> } // namespace fsd
PUSHMI_INLINE_VAR constexpr detail::get_fn<T> get{}; #endif
} // namespace operators
} // namespace pushmi } // namespace pushmi
// clang-format off // clang-format off
...@@ -9507,6 +9647,112 @@ auto via(ExecutorFactory factory) { ...@@ -9507,6 +9647,112 @@ auto via(ExecutorFactory factory) {
} // namespace fsd } // namespace fsd
#endif #endif
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include "../single.h"
//#include "submit.h"
//#include "extension_operators.h"
//#include "via.h"
namespace pushmi {
namespace detail {
struct request_via_fn {
template<typename In>
struct semisender {
In in;
template<class... AN>
auto via(AN&&... an) {
return in | ::pushmi::operators::via((AN&&) an...);
}
};
auto operator()() const;
};
auto request_via_fn::operator()() const {
return constrain(lazy::Sender<_1>, [](auto in) {
using In = decltype(in);
return semisender<In>{in};
});
}
} // namespace detail
namespace operators {
PUSHMI_INLINE_VAR constexpr detail::request_via_fn request_via{};
} // namespace operators
PUSHMI_TEMPLATE(class To, class In)
(requires Same<To, is_sender<>> && Sender<_1>)
auto via_cast(In in) {
return in;
}
PUSHMI_TEMPLATE(class To, class In)
(requires Same<To, is_sender<>>)
auto via_cast(detail::request_via_fn::semisender<In> ss) {
return ss.in;
}
} // namespace pushmi
// clang-format off
// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
//#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
//#include "../single.h"
//#include "submit.h"
//#include "extension_operators.h"
//#include "../subject.h"
namespace pushmi {
namespace detail {
template<class T>
struct share_fn {
auto operator()() const;
};
template<class T>
auto share_fn<T>::operator()() const {
return constrain(lazy::Sender<_1>, [](auto in) {
using In = decltype(in);
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());
));
return sub;
});
}
} // namespace detail
namespace operators {
template<class T>
PUSHMI_INLINE_VAR constexpr detail::share_fn<T> share{};
} // namespace operators
} // namespace pushmi } // namespace pushmi
#endif // PUSHMI_SINGLE_HEADER #endif // PUSHMI_SINGLE_HEADER
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
// This source code is licensed under the MIT license found in the // This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree. // LICENSE file in the root directory of this source tree.
#include <vector>
#include <pushmi/time_single_deferred.h> #include <pushmi/time_single_deferred.h>
namespace pushmi { namespace pushmi {
......
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