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

change copyright headers and apply some clang format

Summary:
changed the copyright headers to match folly
ran clang format in places that it did not mess up too badly

almost entirely whitespace changes

Reviewed By: Orvid

Differential Revision: D10450516

fbshipit-source-id: dc00a3952ee63d9ceb218aacb8f5f55a7c171170
parent a25bbf7c
#define NONIUS_RUNNER
#define concept Concept
#include <nonius/nonius.h++>
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <chrono>
#include <cstdint>
......
This diff is collapsed.
#pragma once
// Copyright (c) 2018-present, Facebook, Inc.
//
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/receiver.h>
#include <folly/experimental/pushmi/executor.h>
......
// clang-format off
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <type_traits>
......@@ -401,7 +411,8 @@ PUSHMI_PP_IGNORE_CXX2A_COMPAT_BEGIN
#else
#define PUSHMI_BROKEN_SUBSUMPTION(...) __VA_ARGS__
#define PUSHMI_TYPE_CONSTRAINT(...) class
// bool() is used to prevent 'error: pasting "PUSHMI_PP_REQUIRES_PROBE_" and "::" does not give a valid preprocessing token'
// bool() is used to prevent 'error: pasting "PUSHMI_PP_REQUIRES_PROBE_" and
// "::" does not give a valid preprocessing token'
#define PUSHMI_EXP(...) bool(::pushmi::expAnd(__VA_ARGS__))
#define PUSHMI_AND ,
#endif
......@@ -439,91 +450,91 @@ struct Or;
template <class T>
struct Not {
explicit constexpr operator bool() const noexcept {
return !(bool) T{};
}
PUSHMI_TEMPLATE (class This = Not, bool B)
(requires B == (bool) This{})
constexpr operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return T{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<Not, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<Not, That>{};
}
explicit constexpr operator bool() const noexcept {
return !(bool)T{};
}
PUSHMI_TEMPLATE(class This = Not, bool B)
(requires B == (bool)This{}) constexpr
operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return T{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<Not, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<Not, That>{};
}
};
template <class T, class U>
struct And {
explicit constexpr operator bool() const noexcept {
return (bool) std::conditional_t<(bool) T{}, U, std::false_type>{};
}
PUSHMI_TEMPLATE (class This = And, bool B)
(requires B == (bool) This{})
constexpr operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return Not<And>{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<And, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<And, That>{};
}
explicit constexpr operator bool() const noexcept {
return (bool)std::conditional_t<(bool)T{}, U, std::false_type>{};
}
PUSHMI_TEMPLATE(class This = And, bool B)
(requires B == (bool)This{}) constexpr
operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return Not<And>{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<And, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<And, That>{};
}
};
template <class T, class U>
struct Or {
explicit constexpr operator bool() const noexcept {
return (bool) std::conditional_t<(bool) T{}, std::true_type, U>{};
}
PUSHMI_TEMPLATE (class This = Or, bool B)
(requires B == (bool) This{})
constexpr operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return Not<Or>{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<Or, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<Or, That>{};
}
explicit constexpr operator bool() const noexcept {
return (bool)std::conditional_t<(bool)T{}, std::true_type, U>{};
}
PUSHMI_TEMPLATE(class This = Or, bool B)
(requires B == (bool)This{}) constexpr
operator std::integral_constant<bool, B>() const noexcept {
return {};
}
constexpr auto operator!() const noexcept {
return Not<Or>{};
}
template <class That>
constexpr auto operator&&(That) const noexcept {
return And<Or, That>{};
}
template <class That>
constexpr auto operator||(That) const noexcept {
return Or<Or, That>{};
}
};
} // namespace detail
} // namespace concepts
namespace isolated {
template<class T0>
template <class T0>
constexpr auto expAnd(T0&& t0) {
return (T0&&)t0;
return (T0 &&) t0;
}
template<class T0, class... TN>
template <class T0, class... TN>
constexpr auto expAnd(T0&&, TN&&... tn) {
return concepts::detail::And<T0, decltype(isolated::expAnd((TN&&)tn...))>{};
return concepts::detail::And<T0, decltype(isolated::expAnd((TN &&) tn...))>{};
}
}
} // namespace isolated
template<class... TN>
template <class... TN>
constexpr auto expAnd(TN&&... tn) {
return isolated::expAnd((TN&&)tn...);
return isolated::expAnd((TN &&) tn...);
}
template <class T>
......@@ -532,8 +543,7 @@ constexpr bool implicitly_convertible_to(T) {
}
#ifdef __clang__
template <bool B>
std::enable_if_t<B> requires_()
{}
std::enable_if_t<B> requires_() {}
#else
template <bool B>
PUSHMI_INLINE_VAR constexpr std::enable_if_t<B, int> requires_ = 0;
......
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <functional>
......@@ -13,27 +22,28 @@
namespace pushmi {
PUSHMI_INLINE_VAR constexpr struct invoke_fn {
private:
private:
template <class F>
using mem_fn_t = decltype(std::mem_fn(std::declval<F>()));
public:
public:
template <class F, class... As>
auto operator()(F&& f, As&&...as) const
noexcept(noexcept(((F&&) f)((As&&) as...))) ->
decltype(((F&&) f)((As&&) as...)) {
return ((F&&) f)((As&&) as...);
auto operator()(F&& f, As&&... as) const
noexcept(noexcept(((F &&) f)((As &&) as...)))
-> decltype(((F &&) f)((As &&) as...)) {
return ((F &&) f)((As &&) as...);
}
template <class F, class... As>
auto operator()(F&& f, As&&...as) const
noexcept(noexcept(std::declval<mem_fn_t<F>>()((As&&) as...))) ->
decltype(std::mem_fn(f)((As&&) as...)) {
return std::mem_fn(f)((As&&) as...);
auto operator()(F&& f, As&&... as) const
noexcept(noexcept(std::declval<mem_fn_t<F>>()((As &&) as...)))
-> decltype(std::mem_fn(f)((As &&) as...)) {
return std::mem_fn(f)((As &&) as...);
}
} invoke {};
} invoke{};
template <class F, class...As>
template <class F, class... As>
using invoke_result_t =
decltype(pushmi::invoke(std::declval<F>(), std::declval<As>()...));
decltype(pushmi::invoke(std::declval<F>(), std::declval<As>()...));
PUSHMI_CONCEPT_DEF(
template (class F, class... Args)
......
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// Usage:
......@@ -145,41 +154,42 @@ namespace detail {
template <bool>
struct select {
template <class R, class = std::enable_if_t<!std::is_void<R>::value>>
struct eat_return {
R value_;
template <class T>
constexpr R operator->*(T&&) {
return static_cast<R&&>(value_);
}
};
struct eat {
template <class T>
constexpr void operator->*(T&&) {}
};
template <class R, class = std::enable_if_t<!std::is_void<R>::value>>
struct eat_return {
R value_;
template <class T>
constexpr auto operator->*(T&& t) -> eat_return<decltype(t(::pushmi::detail::id_fn{}))> {
return {t(::pushmi::detail::id_fn{})};
constexpr R operator->*(T&&) {
return static_cast<R&&>(value_);
}
};
struct eat {
template <class T>
constexpr auto operator->*(T&& t) const -> eat {
return t(::pushmi::detail::id_fn{}), void(), eat{};
}
constexpr void operator->*(T&&) {}
};
template <class T>
constexpr auto operator->*(T&& t)
-> eat_return<decltype(t(::pushmi::detail::id_fn{}))> {
return {t(::pushmi::detail::id_fn{})};
}
template <class T>
constexpr auto operator->*(T&& t) const -> eat {
return t(::pushmi::detail::id_fn{}), void(), eat{};
}
};
template <>
struct select<false> {
struct eat {
template <class T>
constexpr auto operator->*(T&& t) -> decltype(auto) {
return t(::pushmi::detail::id_fn{});
}
};
struct eat {
template <class T>
constexpr eat operator->*(T&&) {
return {};
constexpr auto operator->*(T&& t) -> decltype(auto) {
return t(::pushmi::detail::id_fn{});
}
};
template <class T>
constexpr eat operator->*(T&&) {
return {};
}
};
}
}
} // namespace detail
} // namespace pushmi
#endif
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#if __cpp_lib_optional >= 201606
#include <optional> // @manual
......@@ -16,18 +25,18 @@ namespace detail {
#if __cpp_lib_optional >= 201606
template <class T>
struct opt : private std::optional<T> {
opt() = default;
opt& operator=(T&& t) {
this->std::optional<T>::operator=(std::move(t));
return *this;
}
using std::optional<T>::operator*;
using std::optional<T>::operator bool;
opt() = default;
opt& operator=(T&& t) {
this->std::optional<T>::operator=(std::move(t));
return *this;
}
using std::optional<T>::operator*;
using std::optional<T>::operator bool;
};
#else
template <class T>
struct opt {
private:
private:
bool empty_ = true;
std::aligned_union_t<0, T> data_;
T* ptr() {
......@@ -42,40 +51,43 @@ private:
empty_ = true;
}
}
public:
public:
opt() = default;
opt(T&& t) noexcept(std::is_nothrow_move_constructible<T>::value) {
::new(ptr()) T(std::move(t));
::new (ptr()) T(std::move(t));
empty_ = false;
}
opt(const T& t) {
::new(ptr()) T(t);
::new (ptr()) T(t);
empty_ = false;
}
opt(opt&& that) noexcept(std::is_nothrow_move_constructible<T>::value) {
if (that) {
::new(ptr()) T(std::move(*that));
::new (ptr()) T(std::move(*that));
empty_ = false;
that.reset();
}
}
opt(const opt& that) {
if (that) {
::new(ptr()) T(*that);
::new (ptr()) T(*that);
empty_ = false;
}
}
~opt() { reset(); }
opt& operator=(opt&& that)
noexcept(std::is_nothrow_move_constructible<T>::value &&
std::is_nothrow_move_assignable<T>::value) {
~opt() {
reset();
}
opt& operator=(opt&& that) noexcept(
std::is_nothrow_move_constructible<T>::value&&
std::is_nothrow_move_assignable<T>::value) {
if (*this && that) {
**this = std::move(*that);
that.reset();
} else if (*this) {
reset();
} else if (that) {
::new(ptr()) T(std::move(*that));
::new (ptr()) T(std::move(*that));
empty_ = false;
}
return *this;
......@@ -86,17 +98,18 @@ public:
} else if (*this) {
reset();
} else if (that) {
::new(ptr()) T(*that);
::new (ptr()) T(*that);
empty_ = false;
}
return *this;
}
opt& operator=(T&& t) noexcept(std::is_nothrow_move_constructible<T>::value &&
std::is_nothrow_move_assignable<T>::value) {
opt& operator=(T&& t) noexcept(
std::is_nothrow_move_constructible<T>::value&&
std::is_nothrow_move_assignable<T>::value) {
if (*this)
**this = std::move(t);
else {
::new(ptr()) T(std::move(t));
::new (ptr()) T(std::move(t));
empty_ = false;
}
return *this;
......@@ -105,7 +118,7 @@ public:
if (*this)
**this = t;
else {
::new(ptr()) T(t);
::new (ptr()) T(t);
empty_ = false;
}
return *this;
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/forwards.h>
namespace pushmi {
// template <class T, class Dual>
// struct entangled {
// T t;
......@@ -84,7 +94,11 @@ struct entangled {
// have to hold *both* locks to write any of either entangled object's
// metadata, but need only one to read it.
int expected = kUnlocked;
if (!stateMachine.compare_exchange_weak(expected, kLocked, std::memory_order_seq_cst, std::memory_order_relaxed)) {
if (!stateMachine.compare_exchange_weak(
expected,
kLocked,
std::memory_order_seq_cst,
std::memory_order_relaxed)) {
return false;
}
// Having *either* object local-locked protects the data in both objects.
......@@ -93,7 +107,8 @@ struct entangled {
return true;
}
expected = kUnlocked;
if (dual->stateMachine.compare_exchange_strong(expected, kLocked, std::memory_order_seq_cst)) {
if (dual->stateMachine.compare_exchange_strong(
expected, kLocked, std::memory_order_seq_cst)) {
return true;
}
// We got here, and so hit the race; we're deadlocked if we stick to
......@@ -105,7 +120,8 @@ struct entangled {
if ((uintptr_t)this < (uintptr_t)dual) {
// I get to win the race. I'll acquire the locks, but have to make sure
// my memory stays valid until the other thread acknowledges its loss.
while (stateMachine.load(std::memory_order_relaxed) != kLockedAndLossAcknowledged) {
while (stateMachine.load(std::memory_order_relaxed) !=
kLockedAndLossAcknowledged) {
// Spin.
}
stateMachine.store(kLocked, std::memory_order_relaxed);
......@@ -113,7 +129,8 @@ struct entangled {
} else {
// I lose the race, but have to coordinate with the winning thread, so
// that it knows that I'm not about to try to touch it's data
dual->stateMachine.store(kLockedAndLossAcknowledged, std::memory_order_relaxed);
dual->stateMachine.store(
kLockedAndLossAcknowledged, std::memory_order_relaxed);
return false;
}
}
......@@ -194,11 +211,11 @@ struct entangled {
};
template <class First, class Second>
using entangled_pair = std::pair<entangled<First, Second>, entangled<Second, First>>;
using entangled_pair =
std::pair<entangled<First, Second>, entangled<Second, First>>;
template <class First, class Second>
auto entangle(First f, Second s)
-> entangled_pair<First, Second> {
auto entangle(First f, Second s) -> entangled_pair<First, Second> {
entangled<First, Second> ef(std::move(f));
entangled<Second, First> es(std::move(s));
ef.dual = std::addressof(es);
......@@ -209,8 +226,12 @@ auto entangle(First f, Second s)
template <class T, class Dual>
struct locked_entangled_pair : std::pair<T*, Dual*> {
entangled<T, Dual>* e;
~locked_entangled_pair(){if (!!e) {e->unlockBoth();}}
explicit locked_entangled_pair(entangled<T, Dual>& e) : e(std::addressof(e)){
~locked_entangled_pair() {
if (!!e) {
e->unlockBoth();
}
}
explicit locked_entangled_pair(entangled<T, Dual>& e) : e(std::addressof(e)) {
this->e->lockBoth();
this->first = std::addressof(this->e->t);
this->second = !!this->e->dual ? std::addressof(this->e->dual->t) : nullptr;
......@@ -218,9 +239,13 @@ struct locked_entangled_pair : std::pair<T*, Dual*> {
locked_entangled_pair() = delete;
locked_entangled_pair(const locked_entangled_pair&) = delete;
locked_entangled_pair& operator=(const locked_entangled_pair&) = delete;
locked_entangled_pair(locked_entangled_pair&& o) : std::pair<T*, Dual*>(o), e(o.e) {o.e = nullptr;}
locked_entangled_pair& operator=(locked_entangled_pair&& o){
static_cast<std::pair<T*, Dual*>&>(*this) = static_cast<std::pair<T*, Dual*>&&>(o);
locked_entangled_pair(locked_entangled_pair&& o)
: std::pair<T*, Dual*>(o), e(o.e) {
o.e = nullptr;
}
locked_entangled_pair& operator=(locked_entangled_pair&& o) {
static_cast<std::pair<T*, Dual*>&>(*this) =
static_cast<std::pair<T*, Dual*>&&>(o);
e = o.e;
o.e = nullptr;
return *this;
......@@ -228,7 +253,7 @@ struct locked_entangled_pair : std::pair<T*, Dual*> {
};
template <class T, class Dual>
locked_entangled_pair<T, Dual> lock_both(entangled<T, Dual>& e){
locked_entangled_pair<T, Dual> lock_both(entangled<T, Dual>& e) {
return locked_entangled_pair<T, Dual>{e};
}
......@@ -237,27 +262,31 @@ struct shared_entangled : std::shared_ptr<T> {
Dual* dual;
std::mutex* lock;
template<class P>
explicit shared_entangled(std::shared_ptr<P>& p, T& t, Dual& d, std::mutex& l) :
std::shared_ptr<T>(p, std::addressof(t)), dual(std::addressof(d)), lock(std::addressof(l)){
}
template <class P>
explicit shared_entangled(std::shared_ptr<P>& p, T& t, Dual& d, std::mutex& l)
: std::shared_ptr<T>(p, std::addressof(t)),
dual(std::addressof(d)),
lock(std::addressof(l)) {}
shared_entangled() = delete;
};
template <class First, class Second>
using shared_entangled_pair = std::pair<shared_entangled<First, Second>, shared_entangled<Second, First>>;
using shared_entangled_pair =
std::pair<shared_entangled<First, Second>, shared_entangled<Second, First>>;
template <class First, class Second>
auto shared_entangle(First f, Second s)
-> shared_entangled_pair<First, Second> {
struct storage {
storage(First&& f, Second&& s) : p((First&&) f, (Second&&) s) {}
storage(First&& f, Second&& s) : p((First &&) f, (Second &&) s) {}
std::tuple<First, Second> p;
std::mutex lock;
};
auto p = std::make_shared<storage>(std::move(f), std::move(s));
shared_entangled<First, Second> ef(p, std::get<0>(p->p), std::get<1>(p->p), p->lock);
shared_entangled<Second, First> es(p, std::get<1>(p->p), std::get<0>(p->p), p->lock);
shared_entangled<First, Second> ef(
p, std::get<0>(p->p), std::get<1>(p->p), p->lock);
shared_entangled<Second, First> es(
p, std::get<1>(p->p), std::get<0>(p->p), p->lock);
return {std::move(ef), std::move(es)};
}
......@@ -269,7 +298,8 @@ struct locked_shared_entangled_pair : std::pair<T*, Dual*> {
e.lock->unlock();
}
}
explicit locked_shared_entangled_pair(shared_entangled<T, Dual>& e) : e(std::move(e)){
explicit locked_shared_entangled_pair(shared_entangled<T, Dual>& e)
: e(std::move(e)) {
this->e.lock->lock();
this->first = this->e.get();
this->second = this->e.dual;
......@@ -277,9 +307,8 @@ struct locked_shared_entangled_pair : std::pair<T*, Dual*> {
locked_shared_entangled_pair() = delete;
};
template <class T, class Dual>
locked_shared_entangled_pair<T, Dual> lock_both(shared_entangled<T, Dual>& e){
locked_shared_entangled_pair<T, Dual> lock_both(shared_entangled<T, Dual>& e) {
return locked_shared_entangled_pair<T, Dual>{e};
}
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/single_sender.h>
......@@ -12,7 +22,13 @@ namespace pushmi {
namespace operators {
PUSHMI_INLINE_VAR constexpr struct bulk_fn {
template<class F, class ShapeBegin, class ShapeEnd, class Target, class IF, class RS>
template <
class F,
class ShapeBegin,
class ShapeEnd,
class Target,
class IF,
class RS>
auto operator()(
F&& func,
ShapeBegin sb,
......@@ -20,18 +36,28 @@ PUSHMI_INLINE_VAR constexpr struct bulk_fn {
Target&& driver,
IF&& initFunc,
RS&& selector) const {
return [func, sb, se, driver, initFunc, selector](auto in){
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_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));
}
));
});
[in, func, sb, se, driver, initFunc, selector](auto out) mutable {
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));
}));
});
};
}
} bulk {};
} bulk{};
} // namespace operators
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
......@@ -12,91 +27,80 @@
using namespace pushmi::aliases;
struct f_t {};
f_t f(){
f_t f() {
return {};
}
struct g_t {};
g_t g(f_t){
g_t g(f_t) {
return {};
}
// these expressions are read backward, bottom-right to top-left
template<class CPUExecutor, class IOExecutor>
template <class CPUExecutor, class IOExecutor>
void lisp(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
op::submit([](g_t){})(
op::transform([](f_t ft) {return g(ft);})(
op::transform([](auto){ return f(); })(
cpu)));
// 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(mi::strands(cpu))(
op::transform([](auto){ return f(); })(
cpu))));
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions)
op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
op::transform([](auto) { return f(); })(cpu)));
// 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(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(mi::strands(cpu))(
op::transform([](auto){ return f(); })(
io))));
op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(io))));
}
template<class CPUExecutor, class IOExecutor>
template <class CPUExecutor, class IOExecutor>
void sugar(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
cpu |
op::transform([](auto){ return f(); }) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
// 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(mi::strands(cpu)) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions)
cpu | op::transform([](auto) { return f(); }) |
op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
// 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(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(mi::strands(cpu)) |
op::transform([](f_t ft) {return g(ft);}) |
op::submit([](g_t){});
io | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) |
op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
}
template<class CPUExecutor, class IOExecutor>
template <class CPUExecutor, class IOExecutor>
void pipe(CPUExecutor cpu, IOExecutor io) {
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all the functions)
// f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
// the functions)
mi::pipe(
cpu,
op::transform([](auto){ return f(); }),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
cpu,
op::transform([](auto) { return f(); }),
op::transform([](f_t ft) { return g(ft); }),
op::submit([](g_t) {}));
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu task runs g)
// f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
// task runs g)
mi::pipe(
cpu,
op::transform([](auto){ return f(); }),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
cpu,
op::transform([](auto) { return f(); }),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) { return g(ft); }),
op::submit([](g_t) {}));
// f on io - g on cpu
mi::pipe(
io,
op::transform([](auto){ return f(); }),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) {return g(ft);}),
op::submit([](g_t){}));
io,
op::transform([](auto) { return f(); }),
op::via(mi::strands(cpu)),
op::transform([](f_t ft) { return g(ft); }),
op::submit([](g_t) {}));
}
int main()
{
mi::pool cpuPool{std::max(1u,std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool cpuPool{std::max(1u, std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u, std::thread::hardware_concurrency())};
lisp(cpuPool.executor(), ioPool.executor());
sugar(cpuPool.executor(), ioPool.executor());
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/o/defer.h>
#include <folly/experimental/pushmi/o/share.h>
......@@ -20,52 +35,47 @@ using namespace pushmi::aliases;
// this constructs eagerly and submits just() lazily
auto defer_execution() {
printf("construct just\n");
return op::just(42) |
op::tap([](int v){ printf("just - %d\n", v); });
return op::just(42) | op::tap([](int v) { printf("just - %d\n", v); });
}
// this constructs defer() eagerly, constructs just() and submits just() lazily
auto defer_construction() {
return op::defer([]{
return defer_execution();
});
return op::defer([] { return defer_execution(); });
}
// this constructs defer(), constructs just() and submits just() eagerly
auto eager_execution() {
return defer_execution() | op::share<int>();
return defer_execution() | op::share<int>();
}
int main()
{
int main() {
printf("\ncall defer_execution\n");
auto de = defer_execution();
printf("submit defer_execution\n");
de | op::submit();
// call defer_execution
// construct just
// submit defer_execution
// just - 42
// call defer_execution
// construct just
// submit defer_execution
// just - 42
printf("\ncall defer_construction\n");
auto dc = defer_construction();
printf("submit defer_construction\n");
dc | op::submit();
// call defer_construction
// submit defer_construction
// construct just
// just - 42
// call defer_construction
// submit defer_construction
// construct just
// just - 42
printf("\ncall eager_execution\n");
auto ee = eager_execution();
printf("submit eager_execution\n");
ee | op::submit();
// call eager_execution
// construct just
// just - 42
// submit eager_execution
// call eager_execution
// construct just
// just - 42
// submit eager_execution
std::cout << "OK" << std::endl;
// OK
// OK
}
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
......@@ -14,28 +29,26 @@
using namespace pushmi::aliases;
template<class Io>
template <class Io>
auto io_operation(Io io) {
return io |
op::transform([](auto){ return 42; }) |
op::tap([](int v){ printf("io pool producing, %d\n", v); }) |
return io | op::transform([](auto) { return 42; }) |
op::tap([](int v) { printf("io pool producing, %d\n", v); }) |
op::request_via();
}
int main()
{
mi::pool cpuPool{std::max(1u,std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool cpuPool{std::max(1u, std::thread::hardware_concurrency())};
mi::pool ioPool{std::max(1u, std::thread::hardware_concurrency())};
auto io = ioPool.executor();
auto cpu = cpuPool.executor();
io_operation(io).via(mi::strands(cpu)) |
op::tap([](int v){ printf("cpu pool processing, %d\n", v); }) |
op::submit();
op::tap([](int v) { printf("cpu pool processing, %d\n", v); }) |
op::submit();
// when the caller is not going to process the result (only side-effect matters)
// or the caller is just going to push the result into a queue.
// when the caller is not going to process the result (only side-effect
// matters) or the caller is just going to push the result into a queue.
// provide a way to skip the transition to a different executor and make it
// stand out so that it has to be justified in code reviews.
mi::via_cast<mi::is_sender<>>(io_operation(io)) | op::submit();
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/examples/bulk.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/submit.h>
namespace pushmi {
PUSHMI_INLINE_VAR constexpr struct for_each_fn {
private:
private:
template <class Function>
struct fn {
Function f_;
......@@ -23,29 +33,28 @@ private:
};
struct identity {
template <class T>
auto operator()(T&& t) const { return (T&&) t; }
auto operator()(T&& t) const {
return (T &&) t;
}
};
struct zero {
int operator()(detail::any) const noexcept { return 0; }
int operator()(detail::any) const noexcept {
return 0;
}
};
public:
template<class ExecutionPolicy, class RandomAccessIterator, class Function>
public:
template <class ExecutionPolicy, class RandomAccessIterator, class Function>
void operator()(
ExecutionPolicy&& policy,
RandomAccessIterator begin,
RandomAccessIterator end,
Function f) const {
operators::just(0) |
operators::bulk(
fn<Function>{f},
begin,
end,
policy,
identity{},
zero{}
) |
operators::blocking_submit();
operators::bulk(
fn<Function>{f}, begin, end, policy, identity{}, zero{}) |
operators::blocking_submit();
}
} for_each {};
} for_each{};
} // namespace pushmi
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/examples/for_each.h>
#include <folly/experimental/pushmi/examples/pool.h>
using namespace pushmi::aliases;
template<class Executor, class Allocator = std::allocator<char>>
template <class Executor, class Allocator = std::allocator<char>>
auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
return [e, a](
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
using RS = decltype(selector);
using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value,
decltype(func),
typename std::remove_reference<decltype(func)>::type>;
using Out = decltype(out);
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> allocState(a);
auto shared_state = std::allocate_shared<
std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to first exception)
>>(allocState, std::exception_ptr{}, std::move(out), std::move(selector), (decltype(func) &&) func, init(std::move(input)), 1, 0);
e | op::submit([e, sb, se, shared_state](auto){
auto stepDone = [](auto shared_state){
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch(...) {
mi::set_error(std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se; ++idx, ++std::get<5>(*shared_state)){
e | op::submit([shared_state, idx, stepDone](auto ex){
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3>(*shared_state)(step, idx);
} while(!std::get<4>(*shared_state).compare_exchange_strong(old, step));
} catch(...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
using RS = decltype(selector);
using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value,
decltype(func),
typename std::remove_reference<decltype(func)>::type>;
using Out = decltype(out);
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char>
allocState(a);
auto shared_state = std::allocate_shared<std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to
// first exception)
>>(
allocState,
std::exception_ptr{},
std::move(out),
std::move(selector),
(decltype(func)&&)func,
init(std::move(input)),
1,
0);
e | op::submit([e, sb, se, shared_state](auto) {
auto stepDone = [](auto shared_state) {
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(
std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(
std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch (...) {
mi::set_error(
std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se;
++idx, ++std::get<5>(*shared_state)) {
e | op::submit([shared_state, idx, stepDone](auto ex) {
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3> (*shared_state)(step, idx);
} while (!std::get<4>(*shared_state)
.compare_exchange_strong(old, step));
} catch (...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()](auto) mutable {
mi::set_error(out, ep);
});
}
};
stepDone(shared_state);
});
} catch (...) {
e |
op::submit([out = std::move(out), ep = std::current_exception()](
auto) mutable { mi::set_error(out, ep); });
}
};
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::vector<int> vec(10);
mi::for_each(naive_executor_bulk_target(p.executor()), vec.begin(), vec.end(), [](int& x){
x = 42;
});
mi::for_each(
naive_executor_bulk_target(p.executor()),
vec.begin(),
vec.end(),
[](int& x) { x = 42; });
assert(std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
assert(
std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
std::cout << "OK" << std::endl;
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/for_each.h>
using namespace pushmi::aliases;
auto inline_bulk_target() {
return [](
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx){
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch(...) {
mi::set_error(out, std::current_exception());
}
};
return [](auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx) {
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch (...) {
mi::set_error(out, std::current_exception());
}
};
}
int main()
{
int main() {
std::vector<int> vec(10);
mi::for_each(inline_bulk_target(), vec.begin(), vec.end(), [](int& x){
x = 42;
});
mi::for_each(
inline_bulk_target(), vec.begin(), vec.end(), [](int& x) { x = 42; });
assert(std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
assert(
std::count(vec.begin(), vec.end(), 42) == static_cast<int>(vec.size()));
std::cout << "OK" << std::endl;
}
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/single_sender.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/single_sender.h>
namespace pushmi {
namespace detail {
struct no_fail_fn {
private:
private:
struct on_error_impl {
[[noreturn]]
void operator()(any, any) noexcept {
[[noreturn]] void operator()(any, any) noexcept {
std::abort();
}
};
template <class In>
struct out_impl {
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
auto operator()(Out out) const {
(requires Receiver<Out>)auto operator()(Out out) const {
return ::pushmi::detail::receiver_from_fn<In>()(
std::move(out),
::pushmi::on_error(on_error_impl{})
);
std::move(out), ::pushmi::on_error(on_error_impl{}));
}
};
struct in_impl {
PUSHMI_TEMPLATE(class In)
(requires Sender<In>)
auto operator()(In in) const {
(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>{})
);
std::move(in),
::pushmi::detail::submit_transform_out<In>(out_impl<In>{}));
}
};
public:
public:
auto operator()() const {
return in_impl{};
}
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <experimental/thread_pool> // @manual
#include <folly/experimental/pushmi/executor.h>
#include <folly/experimental/pushmi/trampoline.h>
#include <folly/experimental/pushmi/time_single_sender.h>
#include <folly/experimental/pushmi/trampoline.h>
#if __cpp_deduction_guides >= 201703
#define MAKE(x) x MAKE_
#define MAKE_(...) {__VA_ARGS__}
#define MAKE_(...) \
{ __VA_ARGS__ }
#else
#define MAKE(x) make_ ## x
#define MAKE(x) make_##x
#endif
namespace pushmi {
......@@ -23,17 +34,23 @@ namespace pushmi {
using std::experimental::static_thread_pool;
namespace execution = std::experimental::execution;
template<class Executor>
template <class Executor>
struct pool_executor {
using properties = property_set<is_sender<>, is_executor<>, is_never_blocking<>, is_concurrent_sequence<>, 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;
explicit pool_executor(e_t e) : e(std::move(e)) {}
auto executor() { return *this; }
auto executor() {
return *this;
}
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
void submit(Out out) const {
(requires Receiver<Out>)void submit(Out out) const {
e.execute([e = *this, out = std::move(out)]() mutable {
::pushmi::set_value(out, e);
});
......@@ -42,17 +59,22 @@ struct pool_executor {
class pool {
static_thread_pool p;
public:
public:
inline explicit pool(std::size_t threads) : p(threads) {}
inline auto executor() {
auto exec = execution::require(p.executor(), execution::never_blocking, execution::oneway);
auto exec = execution::require(
p.executor(), execution::never_blocking, execution::oneway);
return pool_executor<decltype(exec)>{exec};
}
inline void stop() {p.stop();}
inline void wait() {p.wait();}
inline void stop() {
p.stop();
}
inline void wait() {
p.wait();
}
};
} // 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/examples/bulk.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/submit.h>
namespace pushmi {
PUSHMI_INLINE_VAR constexpr struct reduce_fn {
private:
private:
template <class BinaryOp>
struct fn {
BinaryOp binary_op_;
......@@ -24,28 +34,28 @@ private:
struct identity {
template <class T>
auto operator()(T&& t) const {
return (T&&) t;
return (T &&) t;
}
};
public:
template<class ExecutionPolicy, class ForwardIt, class T, class BinaryOp>
public:
template <class ExecutionPolicy, class ForwardIt, class T, class BinaryOp>
T operator()(
ExecutionPolicy&& policy,
ForwardIt begin,
ForwardIt end,
T init,
BinaryOp binary_op) const {
return operators::just(std::move(init)) |
ExecutionPolicy&& policy,
ForwardIt begin,
ForwardIt end,
T init,
BinaryOp binary_op) const {
return operators::just(std::move(init)) |
operators::bulk(
fn<BinaryOp>{binary_op},
begin,
end,
policy,
identity{},
identity{}
) |
fn<BinaryOp>{binary_op},
begin,
end,
policy,
identity{},
identity{}) |
operators::get<T>;
}
} reduce {};
}
} reduce{};
} // namespace pushmi
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <numeric>
#include <cassert>
#include <iostream>
#include <exception>
#include <iostream>
#include <numeric>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/examples/reduce.h>
using namespace pushmi::aliases;
template<class Executor, class Allocator = std::allocator<char>>
template <class Executor, class Allocator = std::allocator<char>>
auto naive_executor_bulk_target(Executor e, Allocator a = Allocator{}) {
return [e, a](
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
using RS = decltype(selector);
using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value,
decltype(func),
typename std::remove_reference<decltype(func)>::type>;
using Out = decltype(out);
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char> allocState(a);
auto shared_state = std::allocate_shared<
std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to first exception)
>>(allocState, std::exception_ptr{}, std::move(out), std::move(selector), (decltype(func) &&) func, init(std::move(input)), 1, 0);
e | op::submit([e, sb, se, shared_state](auto ){
auto stepDone = [](auto shared_state){
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch(...) {
mi::set_error(std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se; ++idx, ++std::get<5>(*shared_state)){
e | op::submit([shared_state, idx, stepDone](auto ex){
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3>(*shared_state)(step, idx);
} while(!std::get<4>(*shared_state).compare_exchange_strong(old, step));
} catch(...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
using RS = decltype(selector);
using F = std::conditional_t<
std::is_lvalue_reference<decltype(func)>::value,
decltype(func),
typename std::remove_reference<decltype(func)>::type>;
using Out = decltype(out);
try {
typename std::allocator_traits<Allocator>::template rebind_alloc<char>
allocState(a);
auto shared_state = std::allocate_shared<std::tuple<
std::exception_ptr, // first exception
Out, // destination
RS, // selector
F, // func
std::atomic<decltype(init(input))>, // accumulation
std::atomic<std::size_t>, // pending
std::atomic<std::size_t> // exception count (protects assignment to
// first exception)
>>(
allocState,
std::exception_ptr{},
std::move(out),
std::move(selector),
(decltype(func)&&)func,
init(std::move(input)),
1,
0);
e | op::submit([e, sb, se, shared_state](auto) {
auto stepDone = [](auto shared_state) {
// pending
if (--std::get<5>(*shared_state) == 0) {
// first exception
if (std::get<0>(*shared_state)) {
mi::set_error(
std::get<1>(*shared_state), std::get<0>(*shared_state));
return;
}
try {
// selector(accumulation)
auto result = std::get<2>(*shared_state)(
std::move(std::get<4>(*shared_state).load()));
mi::set_value(std::get<1>(*shared_state), std::move(result));
} catch (...) {
mi::set_error(
std::get<1>(*shared_state), std::current_exception());
}
}
};
for (decltype(sb) idx{sb}; idx != se;
++idx, ++std::get<5>(*shared_state)) {
e | op::submit([shared_state, idx, stepDone](auto ex) {
try {
// this indicates to me that bulk is not the right abstraction
auto old = std::get<4>(*shared_state).load();
auto step = old;
do {
step = old;
// func(accumulation, idx)
std::get<3> (*shared_state)(step, idx);
} while (!std::get<4>(*shared_state)
.compare_exchange_strong(old, step));
} catch (...) {
// exception count
if (std::get<6>(*shared_state)++ == 0) {
// store first exception
std::get<0>(*shared_state) = std::current_exception();
} // else eat the exception
}
stepDone(shared_state);
});
} catch(...) {
e | op::submit([out = std::move(out), ep = std::current_exception()](auto) mutable {
mi::set_error(out, ep);
});
}
};
stepDone(shared_state);
});
} catch (...) {
e |
op::submit([out = std::move(out), ep = std::current_exception()](
auto) mutable { mi::set_error(out, ep); });
}
};
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::vector<int> vec(10);
std::fill(vec.begin(), vec.end(), 4);
auto fortyTwo = mi::reduce(naive_executor_bulk_target(p.executor()), vec.begin(), vec.end(), 2, std::plus<>{});
auto fortyTwo = mi::reduce(
naive_executor_bulk_target(p.executor()),
vec.begin(),
vec.end(),
2,
std::plus<>{});
assert(std::accumulate(vec.begin(), vec.end(), 2) == fortyTwo);
......
#include <vector>
#include <numeric>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <cassert>
#include <iostream>
#include <exception>
#include <iostream>
#include <numeric>
#include <vector>
#include <folly/experimental/pushmi/examples/reduce.h>
using namespace pushmi::aliases;
auto inline_bulk_target() {
return [](
auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx){
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch(...) {
mi::set_error(out, std::current_exception());
}
};
return [](auto init,
auto selector,
auto input,
auto&& func,
auto sb,
auto se,
auto out) {
try {
auto acc = init(input);
for (decltype(sb) idx{sb}; idx != se; ++idx) {
func(acc, idx);
}
auto result = selector(std::move(acc));
mi::set_value(out, std::move(result));
} catch (...) {
mi::set_error(out, std::current_exception());
}
};
}
int main()
{
int main() {
std::vector<int> vec(10);
std::fill(vec.begin(), vec.end(), 4);
auto fortyTwo = mi::reduce(inline_bulk_target(), vec.begin(), vec.end(), 2, std::plus<>{});
auto fortyTwo = mi::reduce(
inline_bulk_target(), vec.begin(), vec.end(), 2, std::plus<>{});
assert(std::accumulate(vec.begin(), vec.end(), 2) == fortyTwo);
std::cout << fortyTwo << std::endl;
std::cout << "OK" << std::endl;
}
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/o/empty.h>
#include <folly/experimental/pushmi/o/filter.h>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/tap.h>
#include <folly/experimental/pushmi/o/filter.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/empty.h>
using namespace pushmi::aliases;
const bool setting_exists = false;
auto get_setting() {
return mi::make_single_sender(
[](auto out){
if(setting_exists) {
op::just(42) | op::submit(out);
} else {
op::empty<int>() | op::submit(out);
}
return mi::make_single_sender([](auto out) {
if (setting_exists) {
op::just(42) | op::submit(out);
} else {
op::empty<int>() | op::submit(out);
}
);
});
}
auto println = [](auto v){std::cout << v << std::endl;};
auto println = [](auto v) { std::cout << v << std::endl; };
// concat not yet implemented
template<class T, class E = std::exception_ptr>
auto concat =
[](auto in){
return mi::make_single_sender(
[in](auto out) mutable {
::pushmi::submit(in, mi::make_receiver(out,
[](auto out, auto v){
::pushmi::submit(v, mi::any_receiver<E, T>(out));
}));
});
};
template <class T, class E = std::exception_ptr>
auto concat = [](auto in) {
return mi::make_single_sender([in](auto out) mutable {
::pushmi::submit(in, mi::make_receiver(out, [](auto out, auto v) {
::pushmi::submit(v, mi::any_receiver<E, T>(out));
}));
});
};
int main()
{
get_setting() |
op::transform([](int i){ return std::to_string(i); }) |
op::submit(println);
int main() {
get_setting() | op::transform([](int i) { return std::to_string(i); }) |
op::submit(println);
op::just(42) |
op::filter([](int i){ return i < 42; }) |
op::transform([](int i){ return std::to_string(i); }) |
op::submit(println);
op::just(42) | op::filter([](int i) { return i < 42; }) |
op::transform([](int i) { return std::to_string(i); }) |
op::submit(println);
op::just(42) |
op::transform([](int i) {
if (i < 42) {
return mi::any_single_sender<std::exception_ptr, std::string>{op::empty<std::string>()};
}
return mi::any_single_sender<std::exception_ptr, std::string>{op::just(std::to_string(i))};
}) |
concat<std::string> |
op::just(42) | op::transform([](int i) {
if (i < 42) {
return mi::any_single_sender<std::exception_ptr, std::string>{
op::empty<std::string>()};
}
return mi::any_single_sender<std::exception_ptr, std::string>{
op::just(std::to_string(i))};
}) | concat<std::string> |
op::submit(println);
std::cout << "OK" << std::endl;
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/error.h>
#include <folly/experimental/pushmi/examples/no_fail.h>
#include <folly/experimental/pushmi/o/empty.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/error.h>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/switch_on_error.h>
#include <folly/experimental/pushmi/examples/no_fail.h>
#include <folly/experimental/pushmi/o/transform.h>
using namespace pushmi::aliases;
// concat not yet implemented
template<class T, class E = std::exception_ptr>
auto concat =
[](auto in){
return mi::make_single_sender(
[in](auto out) mutable {
::pushmi::submit(in, mi::make_receiver(out,
[](auto out, auto v){
::pushmi::submit(v, mi::any_receiver<E, T>(out));
}));
});
};
int main()
{
template <class T, class E = std::exception_ptr>
auto concat = [](auto in) {
return mi::make_single_sender([in](auto out) mutable {
::pushmi::submit(in, mi::make_receiver(out, [](auto out, auto v) {
::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(stop_abort);
op::error(std::exception_ptr{}) | op::submit(stop_abort);
op::error(std::errc::argument_list_too_long) |
op::submit(stop_abort);
op::error(std::errc::argument_list_too_long) | op::submit(stop_abort);
// transform an error
// transform an error
op::error(std::errc::argument_list_too_long) |
op::switch_on_error([](auto) noexcept { return op::error(std::exception_ptr{}); }) |
op::submit(stop_abort);
op::error(std::errc::argument_list_too_long) | op::switch_on_error([
](auto) noexcept { return op::error(std::exception_ptr{}); }) |
op::submit(stop_abort);
// use default value if an error occurs
// use default value if an error occurs
op::just(42) |
op::switch_on_error([](auto) noexcept { return op::just(0); }) |
op::submit();
op::switch_on_error([](auto) noexcept { return op::just(0); }) |
op::submit();
// suppress if an error occurs
// suppress if an error occurs
op::error(std::errc::argument_list_too_long) |
op::switch_on_error([](auto) noexcept { return op::empty(); }) |
op::submit();
op::switch_on_error([](auto) noexcept { return op::empty(); }) |
op::submit();
// abort if an error occurs
// abort if an error occurs
op::just(42) |
op::no_fail() |
op::submit();
op::just(42) | op::no_fail() | op::submit();
// transform value to error_
// transform value to error_
op::just(42) |
op::transform([](auto v) {
using r_t = mi::any_single_sender<std::exception_ptr, int>;
if (v < 40) {
return r_t{op::error<int>(std::exception_ptr{})};
} else {
return r_t{op::just(v)};
}
}) |
concat<int> |
op::submit();
// retry on error
op::just(42) | op::transform([](auto v) {
using r_t = mi::any_single_sender<std::exception_ptr, int>;
if (v < 40) {
return r_t{op::error<int>(std::exception_ptr{})};
} else {
return r_t{op::just(v)};
}
}) | concat<int> |
op::submit();
// retry on error
// http.get(ex) |
// op::timeout(ex, 1s) |
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <memory>
#include <utility>
#include <atomic>
#include <thread>
#include <functional>
#include <futures.h>
#include <futures_static_thread_pool.h>
#include <atomic>
#include <functional>
#include <memory>
#include <thread>
#include <utility>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/strand.h>
#include <folly/experimental/pushmi/o/just.h>
#include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/strand.h>
using namespace pushmi::aliases;
struct inline_executor
{
public:
struct inline_executor {
public:
friend bool operator==(
const inline_executor&, const inline_executor&) noexcept { return true; }
const inline_executor&,
const inline_executor&) noexcept {
return true;
}
friend bool operator!=(
const inline_executor&, const inline_executor&) noexcept { return false; }
template<class Function>
void execute(Function f) const noexcept { f(); }
constexpr bool query(std::experimental::execution::oneway_t) { return true; }
constexpr bool query(std::experimental::execution::twoway_t) { return false; }
constexpr bool query(std::experimental::execution::single_t) { return true; }
const inline_executor&,
const inline_executor&) noexcept {
return false;
}
template <class Function>
void execute(Function f) const noexcept {
f();
}
constexpr bool query(std::experimental::execution::oneway_t) {
return true;
}
constexpr bool query(std::experimental::execution::twoway_t) {
return false;
}
constexpr bool query(std::experimental::execution::single_t) {
return true;
}
};
namespace p1054 {
......@@ -43,57 +71,61 @@ namespace p1054 {
// the promise has to exist prior to .then() getting a continuation.
// there must be a shared allocation to connect the promise and future.
template <class T, class Executor>
std::pair<std::experimental::standard_promise<T>, std::experimental::standard_future<T, std::decay_t<Executor>>>
std::pair<
std::experimental::standard_promise<T>,
std::experimental::standard_future<T, std::decay_t<Executor>>>
make_promise_contract(const Executor& e) {
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
}
template<class Executor, class Function, class Future>
std::experimental::standard_future<
std::result_of_t<Function(std::decay_t<typename std::decay_t<Future>::value_type>&&)>, std::decay_t<Executor>>
then_execute(Executor&& e, Function&& f, Future&& pred) {
using V = std::decay_t<typename std::decay_t<Future>::value_type>;
using T = std::result_of_t<Function(V&&)>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
((Future&&)pred).then([e, p, f](V v) mutable {
e.execute([p, f, v]() mutable {
p.set_value(f(v));
});
return 0;
});
return r;
}
template <class Executor, class Function, class Future>
std::experimental::standard_future<
std::result_of_t<
Function(std::decay_t<typename std::decay_t<Future>::value_type>&&)>,
std::decay_t<Executor>>
then_execute(Executor&& e, Function&& f, Future&& pred) {
using V = std::decay_t<typename std::decay_t<Future>::value_type>;
using T = std::result_of_t<Function(V &&)>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
((Future &&) pred).then([e, p, f](V v) mutable {
e.execute([p, f, v]() mutable { p.set_value(f(v)); });
return 0;
});
return r;
}
} // namespace p1054
namespace p1055 {
template<class Executor, class Function, class Future>
template <class Executor, class Function, class Future>
auto then_execute(Executor&& e, Function&& f, Future&& pred) {
return pred | op::via(mi::strands(e)) | op::transform([f](auto v){return f(v);});
return pred | op::via(mi::strands(e)) |
op::transform([f](auto v) { return f(v); });
}
} // namespace p1055
}
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::experimental::futures_static_thread_pool sp{std::max(1u,std::thread::hardware_concurrency())};
std::experimental::futures_static_thread_pool sp{
std::max(1u, std::thread::hardware_concurrency())};
auto pc = p1054::make_promise_contract<int>(inline_executor{});
auto& pr = std::get<0>(pc);
auto& r = std::get<1>(pc);
auto f = p1054::then_execute(sp.executor(), [](int v){return v*2;}, std::move(r));
auto f = p1054::then_execute(
sp.executor(), [](int v) { return v * 2; }, std::move(r));
pr.set_value(42);
f.get();
p1055::then_execute(p.executor(), [](int v){return v*2;}, op::just(21)) | op::get<int>;
p1055::then_execute(p.executor(), [](int v) { return v * 2; }, op::just(21)) |
op::get<int>;
sp.stop();
sp.wait();
......
#include <vector>
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <memory>
#include <utility>
#include <futures.h>
#include <atomic>
#include <thread>
#include <functional>
#include <futures.h>
#include <memory>
#include <thread>
#include <utility>
#include <folly/experimental/pushmi/examples/pool.h>
......@@ -25,43 +40,47 @@ namespace p1054 {
// the promise has to exist prior to .then() getting a continuation.
// there must be a shared allocation to connect the promise and future.
template <class T, class Executor>
std::pair<std::experimental::standard_promise<T>, std::experimental::standard_future<T, std::decay_t<Executor>>>
std::pair<
std::experimental::standard_promise<T>,
std::experimental::standard_future<T, std::decay_t<Executor>>>
make_promise_contract(const Executor& e) {
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
std::experimental::standard_promise<T> promise;
auto ex = e;
return {promise, promise.get_future(std::move(ex))};
}
template<class Executor, class Function>
std::experimental::standard_future<std::result_of_t<std::decay_t<Function>()>, std::decay_t<Executor>>
twoway_execute(Executor&& e, Function&& f) {
using T = std::result_of_t<std::decay_t<Function>()>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
e.execute([p, f]() mutable {p.set_value(f());});
return r;
}
template <class Executor, class Function>
std::experimental::standard_future<
std::result_of_t<std::decay_t<Function>()>,
std::decay_t<Executor>>
twoway_execute(Executor&& e, Function&& f) {
using T = std::result_of_t<std::decay_t<Function>()>;
auto pc = make_promise_contract<T>(e);
auto p = std::get<0>(pc);
auto r = std::get<1>(pc);
e.execute([p, f]() mutable { p.set_value(f()); });
return r;
}
} // namespace p1054
namespace p1055 {
template<class Executor, class Function>
template <class Executor, class Function>
auto twoway_execute(Executor&& e, Function&& f) {
return e | op::transform([f](auto){return f();});
return e | op::transform([f](auto) { return f(); });
}
}
} // namespace p1055
int main()
{
mi::pool p{std::max(1u,std::thread::hardware_concurrency())};
int main() {
mi::pool p{std::max(1u, std::thread::hardware_concurrency())};
std::experimental::static_thread_pool sp{std::max(1u,std::thread::hardware_concurrency())};
std::experimental::static_thread_pool sp{
std::max(1u, std::thread::hardware_concurrency())};
p1054::twoway_execute(sp.executor(), [](){return 42;}).get();
p1054::twoway_execute(sp.executor(), []() { return 42; }).get();
p1055::twoway_execute(p.executor(), [](){return 42;}) | op::get<int>;
p1055::twoway_execute(p.executor(), []() { return 42; }) | op::get<int>;
sp.stop();
sp.wait();
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <chrono>
#include <functional>
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <future>
#include <functional>
#include <folly/experimental/pushmi/traits.h>
#include <folly/experimental/pushmi/forwards.h>
#include <folly/experimental/pushmi/properties.h>
#include <folly/experimental/pushmi/traits.h>
namespace pushmi {
namespace __adl {
......@@ -349,7 +361,8 @@ PUSHMI_INLINE_VAR constexpr __adl::get_top_fn now{};
PUSHMI_INLINE_VAR constexpr __adl::get_top_fn top{};
template<class T>
struct property_set_traits<T, std::enable_if_t<(bool)Invocable<T&> && not Valid<T&, __properties_t>>> {
struct property_set_traits<T, std::enable_if_t<(bool)Invocable<T&> &&
not Valid<T&, __properties_t>>> {
using properties = property_set<is_receiver<>>;
};
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/flow_receiver.h>
#include <folly/experimental/pushmi/executor.h>
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/receiver.h>
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/flow_receiver.h>
#include <folly/experimental/pushmi/executor.h>
......@@ -27,12 +38,14 @@ class any_flow_single_sender {
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&, any_flow_receiver<PE, std::ptrdiff_t, E, VN...>) = 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_;
template <class Wrapped>
any_flow_single_sender(Wrapped obj, std::false_type) : any_flow_single_sender() {
any_flow_single_sender(Wrapped obj, std::false_type)
: any_flow_single_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
......@@ -40,9 +53,12 @@ class any_flow_single_sender {
delete static_cast<Wrapped const*>(src.pobj_);
}
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
return any_executor<E>{
::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, any_flow_receiver<PE, std::ptrdiff_t, E, VN...> 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));
}
};
......@@ -56,17 +72,19 @@ class any_flow_single_sender {
struct s {
static void op(data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
new (dst->buffer_)
Wrapped(std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
return any_executor<E>{
::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, any_flow_receiver<PE, std::ptrdiff_t, E, VN...> 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));
*static_cast<Wrapped*>((void*)src.buffer_), std::move(out));
}
};
static const vtable vtbl{s::op, s::executor, s::submit};
......@@ -132,14 +150,19 @@ class flow_single_sender<SF, EXF> {
}
};
template <PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data, class DSF, class DEXF>
template <
PUSHMI_TYPE_CONSTRAINT(Sender<is_single<>, is_flow<>>) Data,
class DSF,
class DEXF>
class flow_single_sender<Data, DSF, DEXF> {
Data data_;
DSF sf_;
DEXF exf_;
public:
using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_flow<>, is_single<>>>;
using properties = property_set_insert_t<
properties_t<Data>,
property_set<is_sender<>, is_flow<>, is_single<>>>;
constexpr flow_single_sender() = default;
constexpr explicit flow_single_sender(Data data)
......@@ -177,7 +200,8 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn {
return flow_single_sender<SF, trampolineEXF>{std::move(sf)};
}
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
(requires True<> && Invocable<EXF&>
PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
auto operator()(SF sf, EXF exf) const {
return flow_single_sender<SF, EXF>{std::move(sf), std::move(exf)};
}
......@@ -194,7 +218,8 @@ PUSHMI_INLINE_VAR constexpr struct make_flow_single_sender_fn {
PUSHMI_TEMPLATE(class Data, class DSF, class DEXF)
(requires Sender<Data, is_single<>, is_flow<>> && Invocable<DEXF&, Data&>)
auto operator()(Data d, DSF sf, DEXF exf) const {
return flow_single_sender<Data, DSF, DEXF>{std::move(d), std::move(sf), std::move(exf)};
return flow_single_sender<Data, DSF, DEXF>{std::move(d),
std::move(sf), std::move(exf)};
}
} const make_flow_single_sender {};
......@@ -208,7 +233,8 @@ PUSHMI_TEMPLATE(class SF)
flow_single_sender(SF) -> flow_single_sender<SF, trampolineEXF>;
PUSHMI_TEMPLATE(class SF, class EXF)
(requires True<> && Invocable<EXF&> PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
(requires True<> && Invocable<EXF&>
PUSHMI_BROKEN_SUBSUMPTION(&& not Sender<SF>))
flow_single_sender(SF, EXF) -> flow_single_sender<SF, EXF>;
PUSHMI_TEMPLATE(class Data)
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <exception>
#include <chrono>
#include <folly/experimental/pushmi/traits.h>
#include <chrono>
#include <exception>
namespace pushmi {
......@@ -18,45 +29,45 @@ struct property_traits;
template <class T, class = void>
struct property_set_traits;
template<class... PropertyN>
template <class... PropertyN>
struct property_set;
// trait & tag types
template<class...TN>
template <class... TN>
struct is_single;
template<class...TN>
template <class... TN>
struct is_many;
template<class...TN>
template <class... TN>
struct is_flow;
template<class...TN>
template <class... TN>
struct is_receiver;
template<class...TN>
template <class... TN>
struct is_sender;
template<class... TN>
template <class... TN>
struct is_executor;
template<class...TN>
template <class... TN>
struct is_time;
template<class...TN>
template <class... TN>
struct is_constrained;
template<class... TN>
template <class... TN>
struct is_always_blocking;
template<class... TN>
template <class... TN>
struct is_never_blocking;
template<class... TN>
template <class... TN>
struct is_maybe_blocking;
template<class... TN>
template <class... TN>
struct is_fifo_sequence;
template<class... TN>
template <class... TN>
struct is_concurrent_sequence;
// implementation types
......@@ -85,95 +96,77 @@ class flow_single_sender;
template <PUSHMI_TYPE_CONSTRAINT(SemiMovable)... TN>
class flow_many_sender;
template <
class E = std::exception_ptr,
class... VN>
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 PE = std::exception_ptr,
class PV = std::ptrdiff_t,
class E = PE,
class... VN>
class any_flow_receiver;
template<
class E = std::exception_ptr,
class... VN>
template <class E = std::exception_ptr, class... VN>
class any_single_sender;
template<
class E = std::exception_ptr,
class... VN>
template <class E = std::exception_ptr, class... VN>
class any_many_sender;
template <
class PE=std::exception_ptr,
class E=PE,
class... VN>
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 PE = std::exception_ptr,
class PV = std::ptrdiff_t,
class E = PE,
class... VN>
class any_flow_many_sender;
template<
class E = std::exception_ptr,
class C = std::ptrdiff_t,
class... VN>
template <class E = std::exception_ptr, class C = std::ptrdiff_t, class... VN>
class any_constrained_single_sender;
template<
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point,
class... VN>
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point,
class... VN>
class any_time_single_sender;
template<
class E = std::exception_ptr>
template <class E = std::exception_ptr>
struct any_executor;
template<
class E = std::exception_ptr>
template <class E = std::exception_ptr>
struct any_executor_ref;
template<
class E = std::exception_ptr,
class CV = std::ptrdiff_t>
template <class E = std::exception_ptr, class CV = std::ptrdiff_t>
struct any_constrained_executor;
template<
class E = std::exception_ptr,
class TP = std::ptrdiff_t>
template <class E = std::exception_ptr, class TP = std::ptrdiff_t>
struct any_constrained_executor_ref;
template<
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
struct any_time_executor;
template<
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
template <
class E = std::exception_ptr,
class TP = std::chrono::system_clock::time_point>
struct any_time_executor_ref;
namespace operators {}
namespace extension_operators {}
namespace aliases {
namespace v = ::pushmi;
namespace mi = ::pushmi;
namespace op = ::pushmi::operators;
namespace ep = ::pushmi::extension_operators;
}
namespace v = ::pushmi;
namespace mi = ::pushmi;
namespace op = ::pushmi::operators;
namespace ep = ::pushmi::extension_operators;
} // namespace aliases
namespace detail {
struct any {
template <class T>
constexpr any(T&&) noexcept {}
};
}
struct any {
template <class T>
constexpr any(T&&) noexcept {}
};
} // namespace detail
} // 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/executor.h>
namespace pushmi {
class inline_constrained_executor_t {
public:
using properties = property_set<is_constrained<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
public:
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>)
void submit(CV, Out out) {
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
std::ptrdiff_t top() {
return 0;
}
auto executor() {
return *this;
}
PUSHMI_TEMPLATE(class CV, class Out)
(requires Regular<CV>&& Receiver<Out>)void submit(CV, Out out) {
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
struct inlineConstrainedEXF {
inline_constrained_executor_t operator()(){
inline_constrained_executor_t operator()() {
return {};
}
};
......@@ -35,24 +52,30 @@ inline inline_constrained_executor_t inline_constrained_executor() {
}
class inline_time_executor_t {
public:
using properties = property_set<is_time<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
public:
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>)
void submit(TP tp, Out out) {
std::this_thread::sleep_until(tp);
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
auto top() {
return std::chrono::system_clock::now();
}
auto executor() {
return *this;
}
PUSHMI_TEMPLATE(class TP, class Out)
(requires Regular<TP>&& Receiver<Out>)void submit(TP tp, Out out) {
std::this_thread::sleep_until(tp);
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
struct inlineTimeEXF {
inline_time_executor_t operator()(){
inline_time_executor_t operator()() {
return {};
}
};
......@@ -62,20 +85,26 @@ inline inline_time_executor_t inline_time_executor() {
}
class inline_executor_t {
public:
using properties = property_set<is_sender<>, is_executor<>, is_always_blocking<>, is_fifo_sequence<>, is_single<>>;
public:
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>)
void submit(Out out) {
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
auto executor() {
return *this;
}
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)void submit(Out out) {
::pushmi::set_value(out, *this);
::pushmi::set_done(out);
}
};
struct inlineEXF {
inline_executor_t operator()(){
inline_executor_t operator()() {
return {};
}
};
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/receiver.h>
#include <folly/experimental/pushmi/executor.h>
#include <folly/experimental/pushmi/receiver.h>
#include <folly/experimental/pushmi/trampoline.h>
namespace pushmi {
......@@ -23,13 +34,15 @@ class any_many_sender {
}
struct vtable {
static void s_op(data&, data*) {}
static any_executor<E> s_executor(data&) { return {}; }
static any_executor<E> s_executor(data&) {
return {};
}
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&, any_receiver<E, VN...>) = vtable::s_submit;
};
static constexpr vtable const noop_ {};
static constexpr vtable const noop_{};
vtable const* vptr_ = &noop_;
template <class Wrapped>
any_many_sender(Wrapped obj, std::false_type) : any_many_sender() {
......@@ -40,7 +53,8 @@ class any_many_sender {
delete static_cast<Wrapped const*>(src.pobj_);
}
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
return any_executor<E>{
::pushmi::executor(*static_cast<Wrapped*>(src.pobj_))};
}
static void submit(data& src, any_receiver<E, VN...> out) {
::pushmi::submit(*static_cast<Wrapped*>(src.pobj_), std::move(out));
......@@ -51,17 +65,17 @@ class any_many_sender {
vptr_ = &vtbl;
}
template <class Wrapped>
any_many_sender(Wrapped obj, std::true_type) noexcept
: any_many_sender() {
any_many_sender(Wrapped obj, std::true_type) noexcept : any_many_sender() {
struct s {
static void op(data& src, data* dst) {
if (dst)
new (dst->buffer_) Wrapped(
std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
new (dst->buffer_)
Wrapped(std::move(*static_cast<Wrapped*>((void*)src.buffer_)));
static_cast<Wrapped const*>((void*)src.buffer_)->~Wrapped();
}
static any_executor<E> executor(data& src) {
return any_executor<E>{::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
return any_executor<E>{
::pushmi::executor(*static_cast<Wrapped*>((void*)src.buffer_))};
}
static void submit(data& src, any_receiver<E, VN...> out) {
::pushmi::submit(
......@@ -74,21 +88,24 @@ class any_many_sender {
}
template <class T, class U = std::decay_t<T>>
using wrapped_t =
std::enable_if_t<!std::is_same<U, any_many_sender>::value, U>;
std::enable_if_t<!std::is_same<U, any_many_sender>::value, U>;
public:
using properties = property_set<is_sender<>, is_many<>>;
any_many_sender() = default;
any_many_sender(any_many_sender&& that) noexcept
: any_many_sender() {
any_many_sender(any_many_sender&& that) noexcept : any_many_sender() {
that.vptr_->op_(that.data_, &data_);
std::swap(that.vptr_, vptr_);
}
PUSHMI_TEMPLATE(class Wrapped)
(requires SenderTo<wrapped_t<Wrapped>, any_receiver<E, VN...>, 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{std::move(obj), bool_<insitu<Wrapped>()>{}} {}
~any_many_sender() {
vptr_->op_(data_, nullptr);
}
......@@ -108,7 +125,7 @@ class any_many_sender {
// Class static definitions:
template <class E, class... VN>
constexpr typename any_many_sender<E, VN...>::vtable const
any_many_sender<E, VN...>::noop_;
any_many_sender<E, VN...>::noop_;
template <class SF, class EXF>
class many_sender<SF, EXF> {
......@@ -119,14 +136,16 @@ class many_sender<SF, EXF> {
using properties = property_set<is_sender<>, is_many<>>;
constexpr many_sender() = default;
constexpr explicit many_sender(SF sf)
: sf_(std::move(sf)) {}
constexpr explicit many_sender(SF sf) : sf_(std::move(sf)) {}
constexpr many_sender(SF sf, EXF exf)
: sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(); }
auto executor() {
return exf_();
}
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out> 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));
}
......@@ -139,29 +158,32 @@ class many_sender<Data, DSF, DEXF> {
DEXF exf_;
public:
using properties = property_set_insert_t<properties_t<Data>, property_set<is_sender<>, is_many<>>>;
using properties = property_set_insert_t<
properties_t<Data>,
property_set<is_sender<>, is_many<>>>;
constexpr many_sender() = default;
constexpr explicit many_sender(Data data)
: data_(std::move(data)) {}
constexpr explicit many_sender(Data data) : data_(std::move(data)) {}
constexpr many_sender(Data data, DSF sf)
: data_(std::move(data)), sf_(std::move(sf)) {}
constexpr many_sender(Data data, DSF sf, DEXF exf)
: data_(std::move(data)), sf_(std::move(sf)), exf_(std::move(exf)) {}
auto executor() { return exf_(data_); }
auto executor() {
return exf_(data_);
}
PUSHMI_TEMPLATE(class Out)
(requires PUSHMI_EXP(lazy::Receiver<Out> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
(requires PUSHMI_EXP(
lazy::Receiver<Out> PUSHMI_AND
lazy::Invocable<DSF&, Data&, Out>))
void submit(Out out) {
sf_(data_, std::move(out));
}
};
template <>
class many_sender<>
: public many_sender<ignoreSF, trampolineEXF> {
public:
class many_sender<> : public many_sender<ignoreSF, trampolineEXF> {
public:
many_sender() = default;
};
......@@ -227,5 +249,4 @@ many_sender(Data, DSF, DEXF) -> many_sender<Data, DSF, DEXF>;
template<>
struct construct_deduced<many_sender> : make_many_sender_fn {};
} // 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/executor.h>
#include <folly/experimental/pushmi/trampoline.h>
......@@ -13,11 +24,18 @@ namespace pushmi {
//
struct new_thread_executor {
using properties = property_set<is_sender<>, is_executor<>, is_never_blocking<>, is_concurrent_sequence<>, is_single<>>;
using properties = property_set<
is_sender<>,
is_executor<>,
is_never_blocking<>,
is_concurrent_sequence<>,
is_single<>>;
new_thread_executor executor() { return {}; }
new_thread_executor executor() {
return {};
}
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
(requires Receiver<Out>)
void submit(Out out) {
std::thread t{[out = std::move(out)]() mutable {
auto tr = ::pushmi::trampoline();
......
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/receiver.h>
#include <folly/experimental/pushmi/single_sender.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
namespace pushmi {
namespace operators {
PUSHMI_INLINE_VAR constexpr struct defer_fn {
private:
private:
template <class F>
struct impl {
F f_;
PUSHMI_TEMPLATE(class Data, class Out)
(requires Receiver<Out>)
(requires Receiver<Out>)
void operator()(Data&, Out out) {
auto sender = f_();
::pushmi::submit(sender, std::move(out));
}
};
public:
public:
PUSHMI_TEMPLATE(class F)
(requires Invocable<F&>)
(requires Invocable<F&>)
auto operator()(F f) const {
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 {};
} defer{};
} // namespace operators
......
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/detail/functional.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h>
namespace pushmi {
namespace detail {
struct single_empty_sender_base : single_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_single<>, is_always_blocking<>, is_fifo_sequence<>>;
};
template <class... VN>
struct single_empty_impl {
PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue<Out, VN...>)
void operator()(single_empty_sender_base&, 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<>>;
};
template <class... VN>
struct single_empty_impl {
PUSHMI_TEMPLATE(class Out)
(requires ReceiveValue<Out, VN...>)
void operator()(
single_empty_sender_base&,
Out out) {
::pushmi::set_done(out);
}
};
} // namespace detail
namespace operators {
template <class... VN>
auto empty() {
return make_single_sender(detail::single_empty_sender_base{}, detail::single_empty_impl<VN...>{});
return make_single_sender(
detail::single_empty_sender_base{}, detail::single_empty_impl<VN...>{});
}
inline auto empty() {
return make_single_sender(detail::single_empty_sender_base{}, detail::single_empty_impl<>{});
return make_single_sender(
detail::single_empty_sender_base{}, detail::single_empty_impl<>{});
}
} // namespace operators
......
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h>
namespace pushmi {
namespace detail {
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 E, class... VN>
struct single_error_impl {
E e_;
PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, E> && ReceiveValue<Out, VN...>)
void operator()(single_error_sender_base&, 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 E, class... VN>
struct single_error_impl {
E e_;
PUSHMI_TEMPLATE(class Out)
(requires ReceiveError<Out, E>&& ReceiveValue<Out, VN...>)
void operator()(
single_error_sender_base&,
Out out) {
::pushmi::set_error(out, std::move(e_));
}
};
} // namespace detail
namespace operators {
PUSHMI_TEMPLATE(class... VN, class E)
(requires And<SemiMovable<VN>...> && SemiMovable<E>)
(requires And<SemiMovable<VN>...>&& SemiMovable<E>)
auto error(E e) {
return make_single_sender(detail::single_error_sender_base{}, detail::single_error_impl<E, VN...>{std::move(e)});
return make_single_sender(
detail::single_error_sender_base{},
detail::single_error_impl<E, VN...>{std::move(e)});
}
} // namespace operators
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/piping.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/piping.h>
namespace pushmi {
namespace detail {
struct filter_fn {
private:
private:
template <class In, class Predicate>
struct on_value_impl {
Predicate p_;
PUSHMI_TEMPLATE(class Out, class... VN)
(requires Receiver<Out>)
(requires Receiver<Out>)
void operator()(Out& out, VN&&... vn) const {
if (p_(as_const(vn)...)) {
::pushmi::set_value(out, (VN&&) vn...);
::pushmi::set_value(out, (VN &&) vn...);
}
}
};
......@@ -28,30 +39,30 @@ private:
struct submit_impl {
Predicate p_;
PUSHMI_TEMPLATE(class Out)
(requires Receiver<Out>)
(requires Receiver<Out>)
auto operator()(Out out) const {
return ::pushmi::detail::receiver_from_fn<In>()(
std::move(out),
// copy 'p' to allow multiple calls to submit
on_value_impl<In, Predicate>{p_}
);
std::move(out),
// copy 'p' to allow multiple calls to submit
on_value_impl<In, Predicate>{p_});
}
};
template <class Predicate>
struct adapt_impl {
Predicate p_;
PUSHMI_TEMPLATE(class In)
(requires Sender<In>)
(requires Sender<In>)
auto operator()(In in) const {
return ::pushmi::detail::sender_from(
std::move(in),
::pushmi::detail::submit_transform_out<In>(submit_impl<In, Predicate>{p_})
);
std::move(in),
::pushmi::detail::submit_transform_out<In>(
submit_impl<In, Predicate>{p_}));
}
};
public:
public:
PUSHMI_TEMPLATE(class Predicate)
(requires SemiMovable<Predicate>)
(requires SemiMovable<Predicate>)
auto operator()(Predicate p) const {
return adapt_impl<Predicate>{std::move(p)};
}
......
#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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h>
......@@ -11,62 +22,79 @@ namespace pushmi {
namespace detail {
struct for_each_fn {
private:
template<class... PN>
struct subset { using properties = property_set<PN...>; };
template<class In, class Out>
private:
template <class... PN>
struct subset {
using properties = property_set<PN...>;
};
template <class In, class Out>
struct Pull : Out {
explicit Pull(Out out) : Out(std::move(out)) {}
using properties = property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>;
using properties =
property_set_insert_t<properties_t<Out>, property_set<is_flow<>>>;
std::function<void(std::ptrdiff_t)> pull;
template<class V>
template <class V>
void value(V&& v) {
::pushmi::set_value(static_cast<Out&>(*this), (V&&) v);
::pushmi::set_value(static_cast<Out&>(*this), (V &&) v);
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires Receiver<Up>)
void starting(Up up){
(requires Receiver<Up>)
void starting(Up up) {
pull = [up = std::move(up)](std::ptrdiff_t requested) mutable {
::pushmi::set_value(up, requested);
};
pull(1);
}
PUSHMI_TEMPLATE(class Up)
(requires ReceiveValue<Up>)
void starting(Up){}
(requires ReceiveValue<Up>)
void starting(Up) {}
};
template <class... AN>
struct fn {
std::tuple<AN...> args_;
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Flow<In> && Many<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_single<>>>>()(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)}));
::pushmi::submit(
in,
::pushmi::detail::receiver_from_fn<In>()(
Pull<In, Out>{std::move(out)}));
return in;
}
PUSHMI_TEMPLATE(class In)
(requires Sender<In> && Constrained<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_single<>>>>()(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::top(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;
}
};
public:
public:
template <class... AN>
auto operator()(AN&&... an) const {
return for_each_fn::fn<AN...>{{(AN&&) an...}};
return for_each_fn::fn<AN...>{{(AN &&) an...}};
}
};
} //namespace detail
} // namespace detail
namespace operators {
PUSHMI_INLINE_VAR constexpr detail::for_each_fn for_each{};
} //namespace operartors
} // namespace operators
} //namespace pushmi
} // namespace pushmi
This diff is collapsed.
// 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/single_sender.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/single_sender.h>
namespace pushmi {
namespace operators {
PUSHMI_INLINE_VAR constexpr struct just_fn {
private:
private:
struct sender_base : single_sender<ignoreSF, inlineEXF> {
using properties = property_set<is_sender<>, is_single<>, is_always_blocking<>, is_fifo_sequence<>>;
using properties = property_set<
is_sender<>,
is_single<>,
is_always_blocking<>,
is_fifo_sequence<>>;
};
template <class... VN>
struct impl {
std::tuple<VN...> vn_;
PUSHMI_TEMPLATE (class Out)
(requires ReceiveValue<Out, VN...>)
PUSHMI_TEMPLATE(class Out)
(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::apply(
::pushmi::set_value,
std::tuple_cat(std::tuple<Out&>{out}, std::move(vn_)));
::pushmi::set_done(std::move(out));
}
};
public:
public:
PUSHMI_TEMPLATE(class... VN)
(requires And<SemiMovable<VN>...>)
(requires And<SemiMovable<VN>...>)
auto operator()(VN... vn) const {
return make_single_sender(sender_base{}, impl<VN...>{std::tuple<VN...>{std::move(vn)...}});
return make_single_sender(
sender_base{}, impl<VN...>{std::tuple<VN...>{std::move(vn)...}});
}
} just {};
} just{};
} // 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.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/piping.h>
#include <folly/experimental/pushmi/executor.h>
#include <folly/experimental/pushmi/o/extension_operators.h>
#include <folly/experimental/pushmi/piping.h>
namespace pushmi {
namespace detail {
struct on_fn {
private:
private:
template <class In, class Out>
struct on_value_impl {
In in_;
......@@ -26,12 +37,12 @@ private:
struct out_impl {
ExecutorFactory ef_;
PUSHMI_TEMPLATE(class Out)
(requires SenderTo<In, Out>)
(requires SenderTo<In, Out>)
void operator()(In& in, Out out) const {
auto exec = ef_();
::pushmi::submit(exec,
::pushmi::make_receiver(on_value_impl<In, Out>{in, std::move(out)})
);
::pushmi::submit(
exec,
::pushmi::make_receiver(on_value_impl<In, Out>{in, std::move(out)}));
}
};
template <class In, class TP, class Out>
......@@ -47,34 +58,34 @@ private:
struct time_out_impl {
ExecutorFactory ef_;
PUSHMI_TEMPLATE(class TP, class Out)
(requires TimeSenderTo<In, Out>)
(requires TimeSenderTo<In, Out>)
void operator()(In& in, TP at, Out out) const {
auto exec = ef_();
::pushmi::submit(exec, at,
::pushmi::make_receiver(
time_on_value_impl<In, TP, Out>{in, at, std::move(out)}
)
);
::pushmi::submit(
exec,
at,
::pushmi::make_receiver(
time_on_value_impl<In, TP, Out>{in, at, std::move(out)}));
}
};
template <class ExecutorFactory>
struct in_impl {
ExecutorFactory ef_;
PUSHMI_TEMPLATE (class In)
(requires Sender<In>)
PUSHMI_TEMPLATE(class In)
(requires Sender<In>)
auto operator()(In in) const {
return ::pushmi::detail::sender_from(
std::move(in),
detail::submit_transform_out<In>(
out_impl<In, ExecutorFactory>{ef_},
time_out_impl<In, ExecutorFactory>{ef_}
)
);
std::move(in),
detail::submit_transform_out<In>(
out_impl<In, ExecutorFactory>{ef_},
time_out_impl<In, ExecutorFactory>{ef_}));
}
};
public:
public:
PUSHMI_TEMPLATE(class ExecutorFactory)
(requires Invocable<ExecutorFactory&> && Executor<invoke_result_t<ExecutorFactory&>>)
(requires Invocable<ExecutorFactory&>&&
Executor<invoke_result_t<ExecutorFactory&>>)
auto operator()(ExecutorFactory ef) const {
return in_impl<ExecutorFactory>{std::move(ef)};
}
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/pushmi/o/submit.h>
using namespace pushmi::aliases;
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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