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

add opt-in support for folly executors and event base

Summary:
adds customizations for folly::Executor folly::EventBase and related types.
The customizations make folly::Executor::KeepAlive<T> model the pushmi Executor, TimeExecutor, and Strand concepts as appropriate

Reviewed By: ericniebler

Differential Revision: D15418629

fbshipit-source-id: d74a777ca743b5fb72664d9c1527837c3fbdb3ba
parent 93b9af64
...@@ -24,6 +24,10 @@ ...@@ -24,6 +24,10 @@
#include <folly/Utility.h> #include <folly/Utility.h>
namespace folly { namespace folly {
namespace pushmi {
// derive from this for types that need to find operator|() overloads by ADL
struct folly_pipeorigin {};
} // namespace pushmi
using Func = Function<void()>; using Func = Function<void()>;
...@@ -64,7 +68,7 @@ class Executor { ...@@ -64,7 +68,7 @@ class Executor {
* preserve the original Executor type. * preserve the original Executor type.
*/ */
template <typename ExecutorT = Executor> template <typename ExecutorT = Executor>
class KeepAlive { class KeepAlive : pushmi::folly_pipeorigin {
public: public:
KeepAlive() = default; KeepAlive() = default;
......
/*
* 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.
*/
#pragma once
#include <folly/experimental/pushmi/customizations/ScheduledExecutor.h>
#include <folly/experimental/pushmi/customizations/SequencedExecutor.h>
#include <folly/experimental/pushmi/executor/executor.h>
#include <folly/io/async/EventBase.h>
namespace folly {
namespace pushmi {
template <class T>
struct property_set_traits_disable<
T,
::folly::Executor,
typename std::enable_if<
std::is_base_of<::folly::EventBase, T>::value>::type> : std::true_type {
};
template <class T>
struct property_set_traits_disable<
T,
::folly::SequencedExecutor,
typename std::enable_if<
std::is_base_of<::folly::EventBase, T>::value>::type> : std::true_type {
};
template <class T>
struct property_set_traits_disable<
T,
::folly::ScheduledExecutor,
typename std::enable_if<
std::is_base_of<::folly::EventBase, T>::value>::type> : std::true_type {
};
template <class T>
struct property_set_traits<
::folly::Executor::KeepAlive<T>,
typename std::enable_if<
std::is_base_of<::folly::EventBase, T>::value>::type> {
using properties = property_set<is_fifo_sequence<>>;
};
} // namespace pushmi
} // namespace folly
/*
* 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.
*/
#pragma once
#include <folly/Executor.h>
#include <folly/experimental/pushmi/executor/executor.h>
namespace folly {
template <typename ExecutorT>
class ExecutorTask : public pushmi::single_sender_tag::with_values<
Executor::KeepAlive<ExecutorT>>::no_error,
public pushmi::pipeorigin {
Executor::KeepAlive<ExecutorT> ex_;
public:
explicit ExecutorTask(Executor::KeepAlive<ExecutorT> ex)
: ex_(std::move(ex)) {}
// assume the worst - specialize pushmi::property_set_traits<> to strengthen
using properties = pushmi::property_set<pushmi::is_maybe_blocking<>>;
template <class TaskReceiver>
void submit(TaskReceiver&& out) && {
// capturing ex into a function stored in ex is a circular reference.
// the only way to break the reference is to destroy the function
ExecutorT* ep = ex_.get();
ep->add([ex = std::move(ex_), pout = (TaskReceiver &&) out]() mutable {
pushmi::set_value(pout, std::move(ex));
pushmi::set_done(pout);
});
}
};
// a derived class can disable the traits definitions by adding the following
//
// namespace pushmi {
// template <class T>
// struct property_set_traits_disable<
// T,
// ::folly::Executor,
// typename std::enable_if<
// std::is_base_of<Derived, T>::value>::type> : std::true_type {
// };
// } // namespace pushmi
//
namespace pushmi {
template <class T>
struct property_set_traits<
::folly::Executor::KeepAlive<T>,
typename std::enable_if<
std::is_base_of<::folly::Executor, T>::value &&
!property_set_traits_disable_v<T, ::folly::Executor>>::type> {
// assume the worst - specialize pushmi::property_set_traits<> to strengthen
// these
using properties = property_set<is_concurrent_sequence<>>;
};
} // namespace pushmi
/// create a sender that will enqueue a call to value() and done() of the
/// receiver on this Executor's execution context. Value will be passed a
/// copy of this Executor. This adds support for Executors to compose with
/// other sender/receiver implementations including algorithms
template <
class ExecutorT,
typename std::enable_if<
std::is_base_of<::folly::Executor, ExecutorT>::value,
int>::type = 0>
ExecutorTask<ExecutorT> schedule(::folly::Executor::KeepAlive<ExecutorT>& se) {
return ExecutorTask<ExecutorT>{se};
}
template <
class ExecutorT,
typename std::enable_if<
std::is_base_of<::folly::Executor, ExecutorT>::value,
int>::type = 0>
ExecutorTask<ExecutorT> schedule(::folly::Executor::KeepAlive<ExecutorT>&& se) {
return ExecutorTask<ExecutorT>{std::move(se)};
}
} // namespace folly
/*
* 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.
*/
#pragma once
#include <folly/executors/InlineExecutor.h>
#include <folly/experimental/pushmi/customizations/Executor.h>
#include <folly/experimental/pushmi/executor/executor.h>
namespace folly {
namespace pushmi {
template <>
struct property_set_traits<
::folly::Executor::KeepAlive<::folly::InlineExecutor>> {
using properties = property_set<is_fifo_sequence<>>;
};
template <>
struct property_set_traits<::folly::ExecutorTask<::folly::InlineExecutor>> {
using properties = property_set<is_always_blocking<>>;
};
} // namespace pushmi
} // namespace folly
/*
* 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.
*/
#pragma once
#include <folly/executors/QueuedImmediateExecutor.h>
#include <folly/experimental/pushmi/customizations/Executor.h>
#include <folly/experimental/pushmi/executor/executor.h>
namespace folly {
namespace pushmi {
template <>
struct property_set_traits<
::folly::Executor::KeepAlive<::folly::QueuedImmediateExecutor>> {
using properties = property_set<is_fifo_sequence<>>;
};
template <>
struct property_set_traits<
::folly::ExecutorTask<::folly::QueuedImmediateExecutor>> {
using properties = property_set<is_maybe_blocking<>>;
};
} // namespace pushmi
} // namespace folly
/*
* 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.
*/
#pragma once
#include <folly/executors/ScheduledExecutor.h>
#include <folly/experimental/pushmi/customizations/Executor.h>
#include <folly/experimental/pushmi/executor/executor.h>
#include <folly/experimental/pushmi/executor/time_executor.h>
namespace folly {
template <typename ExecutorT>
class ScheduledExecutorTask : public pushmi::single_sender_tag::with_values<
Executor::KeepAlive<ExecutorT>>::no_error,
public pushmi::pipeorigin {
using TimePoint = typename ScheduledExecutor::TimePoint;
Executor::KeepAlive<ExecutorT> ex_;
TimePoint at_;
public:
explicit ScheduledExecutorTask(
::folly::Executor::KeepAlive<ExecutorT> ex,
TimePoint at)
: ex_(std::move(ex)), at_(at) {}
using properties = pushmi::property_set<pushmi::is_never_blocking<>>;
template <class TaskReceiver>
void submit(TaskReceiver&& out) && {
// capturing ex into a function stored in ex is a circular reference.
// the only way to break the reference is to destroy the function
ExecutorT* ep = ex_.get();
ep->scheduleAt(
Func{[ex = std::move(ex_), pout = (TaskReceiver &&) out]() mutable {
pushmi::set_value(pout, std::move(ex));
pushmi::set_done(pout);
}},
std::move(at_));
}
};
template <
class ExecutorT,
typename std::enable_if<
std::is_base_of<::folly::ScheduledExecutor, ExecutorT>::value,
int>::type = 0>
auto top(::folly::Executor::KeepAlive<ExecutorT>& se) {
return se->now();
}
template <
class ExecutorT,
typename std::enable_if<
std::is_base_of<::folly::ScheduledExecutor, ExecutorT>::value,
int>::type = 0>
auto schedule(
::folly::Executor::KeepAlive<ExecutorT>& se,
typename ScheduledExecutor::TimePoint at) {
return ScheduledExecutorTask<ExecutorT>{se, std::move(at)};
}
template <
class ExecutorT,
typename std::enable_if<
std::is_base_of<::folly::ScheduledExecutor, ExecutorT>::value,
int>::type = 0>
auto schedule(
::folly::Executor::KeepAlive<ExecutorT>&& se,
typename ScheduledExecutor::TimePoint at) {
return ScheduledExecutorTask<ExecutorT>{std::move(se), std::move(at)};
}
// a derived class can disable the traits definitions by adding the following
//
// namespace pushmi {
// template <class T>
// struct property_set_traits_disable<
// T,
// ::folly::ScheduledExecutor,
// typename std::enable_if<
// std::is_base_of<Derived, T>::value>::type> : std::true_type {
// };
// } // namespace pushmi
//
namespace pushmi {
template <class T>
struct property_set_traits_disable<
T,
::folly::Executor,
typename std::enable_if<
std::is_base_of<::folly::ScheduledExecutor, T>::value &&
!property_set_traits_disable_v<T, ::folly::ScheduledExecutor>>::type>
: std::true_type {};
template <class T>
struct property_set_traits<
::folly::Executor::KeepAlive<T>,
typename std::enable_if<
std::is_base_of<::folly::ScheduledExecutor, T>::value &&
!property_set_traits_disable_v<T, ::folly::ScheduledExecutor>>::type> {
using properties = property_set<is_fifo_sequence<>>;
};
} // namespace pushmi
} // namespace folly
/*
* 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.
*/
#pragma once
#include <folly/executors/SequencedExecutor.h>
#include <folly/experimental/pushmi/customizations/Executor.h>
#include <folly/experimental/pushmi/executor/executor.h>
namespace folly {
// a derived class can disable the traits definitions by adding the following
//
// namespace pushmi {
// template <class T>
// struct property_set_traits_disable<
// T,
// ::folly::SequencedExecutor,
// typename std::enable_if<
// std::is_base_of<Derived, T>::value>::type> : std::true_type {
// };
// } // namespace pushmi
//
namespace pushmi {
template <class T>
struct property_set_traits_disable<
T,
::folly::Executor,
typename std::enable_if<
std::is_base_of<::folly::SequencedExecutor, T>::value &&
!property_set_traits_disable_v<T, ::folly::SequencedExecutor>>::type>
: std::true_type {};
template <class T>
struct property_set_traits<
::folly::Executor::KeepAlive<T>,
typename std::enable_if<
std::is_base_of<::folly::SequencedExecutor, T>::value &&
!property_set_traits_disable_v<T, ::folly::SequencedExecutor>>::type> {
using properties = property_set<is_fifo_sequence<>>;
};
template <class T>
struct property_set_traits<
::folly::ExecutorTask<T>,
typename std::enable_if<
std::is_base_of<::folly::SequencedExecutor, T>::value &&
!property_set_traits_disable_v<T, ::folly::SequencedExecutor>>::type> {
using properties = property_set<is_maybe_blocking<>>;
};
} // namespace pushmi
} // namespace folly
...@@ -24,7 +24,7 @@ PUSHMI_TEMPLATE_DEBUG(class In, class Op) ...@@ -24,7 +24,7 @@ PUSHMI_TEMPLATE_DEBUG(class In, class Op)
(requires PUSHMI_EXP(lazy::Invocable<Op, In>)) // (requires PUSHMI_EXP(lazy::Invocable<Op, In>)) //
decltype(auto) decltype(auto)
operator|(In&& in, Op&& op) { operator|(In&& in, Op&& op) {
return ((Op &&)op)((In &&) in); return ((Op &&) op)((In &&) in);
} }
PUSHMI_INLINE_VAR constexpr struct pipe_fn { PUSHMI_INLINE_VAR constexpr struct pipe_fn {
......
/*
* Copyright 2014-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/customizations/Executor.h>
#include <folly/experimental/pushmi/customizations/EventBase.h>
#include <folly/experimental/pushmi/customizations/InlineExecutor.h>
#include <folly/experimental/pushmi/customizations/QueuedImmediateExecutor.h>
#include <folly/experimental/pushmi/o/for_each.h>
#include <folly/experimental/pushmi/o/from.h>
#include <folly/experimental/pushmi/o/on.h>
#include <folly/experimental/pushmi/o/schedule.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h>
#include <folly/portability/GTest.h>
using namespace folly;
struct ReceiverRecorder {
using receiver_category = folly::pushmi::receiver_tag;
template <class... VN>
void value(VN&&...) {
++values;
}
template <class E>
void error(E&&) noexcept {
++errors;
}
void done() {
++dones;
}
int values{0};
int errors{0};
int dones{0};
};
constexpr class Recorder {
template <class Out>
struct receiver {
using receiver_category = folly::pushmi::receiver_category_t<Out>;
ReceiverRecorder* r_;
std::decay_t<Out> out_;
template <class... VN>
void value(VN&&... vn) {
++r_->values;
folly::pushmi::set_value(out_, (VN &&) vn...);
}
template <class E>
void error(E&& e) noexcept {
++r_->errors;
folly::pushmi::set_error(out_, (E &&) e);
}
void done() {
++r_->dones;
folly::pushmi::set_done(out_);
}
};
template <class In>
struct sender {
ReceiverRecorder* r_;
std::decay_t<In> in_;
using sender_category = folly::pushmi::sender_category_t<In>;
using properties =
folly::pushmi::property_set<folly::pushmi::is_maybe_blocking<>>;
template <class Out>
void submit(Out&& out) & {
folly::pushmi::submit(in_, receiver<Out>{r_, (Out &&) out});
}
template <class Out>
void submit(Out&& out) && {
folly::pushmi::submit(std::move(in_), receiver<Out>{r_, (Out &&) out});
}
};
struct adapt_fn {
ReceiverRecorder* r_;
template <class In>
auto operator()(In&& in) {
return sender<In>{r_, (In &&) in};
}
};
public:
auto operator()(ReceiverRecorder* r) const {
return adapt_fn{r};
}
} recorder;
struct FlowReceiverRecorder {
using receiver_category = folly::pushmi::flow_receiver_tag;
template <class... VN>
void value(VN&&...) {
++values;
}
template <class E>
void error(E&&) noexcept {
++errors;
}
void done() {
++dones;
}
template <class Up>
void starting(Up&&) noexcept {
++startings;
}
int values{0};
int errors{0};
int dones{0};
int startings{0};
};
constexpr class FlowRecorder {
template <class Out>
struct receiver {
using receiver_category = folly::pushmi::receiver_category_t<Out>;
FlowReceiverRecorder* r_;
std::decay_t<Out> out_;
template <class... VN>
void value(VN&&... vn) {
++r_->values;
folly::pushmi::set_value(out_, (VN &&) vn...);
}
template <class E>
void error(E&& e) noexcept {
++r_->errors;
folly::pushmi::set_error(out_, (E &&) e);
}
void done() {
++r_->dones;
folly::pushmi::set_done(out_);
}
template <class Up>
void starting(Up&& up) noexcept {
++r_->startings;
folly::pushmi::set_starting(out_, (Up &&) up);
}
};
template <class In>
struct sender {
FlowReceiverRecorder* r_;
std::decay_t<In> in_;
using sender_category = folly::pushmi::sender_category_t<In>;
using properties = folly::pushmi::properties_t<In>;
template <class Out>
void submit(Out&& out) & {
folly::pushmi::submit(in_, receiver<Out>{r_, (Out &&) out});
}
template <class Out>
void submit(Out&& out) && {
folly::pushmi::submit(std::move(in_), receiver<Out>{r_, (Out &&) out});
}
};
struct adapt_fn {
FlowReceiverRecorder* r_;
template <class In>
auto operator()(In&& in) {
return sender<In>{r_, (In &&) in};
}
};
public:
auto operator()(FlowReceiverRecorder* r) const {
return adapt_fn{r};
}
} flow_recorder;
TEST(Executor, ScheduleInlineExecutor) {
InlineExecutor x;
auto xt = getKeepAliveToken(x);
ReceiverRecorder r{};
auto s = xt | folly::pushmi::operators::schedule() | recorder(&r);
EXPECT_EQ(r.values, 0);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 0);
folly::pushmi::submit(std::move(s), folly::pushmi::receiver<>{});
EXPECT_EQ(r.values, 1);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 1);
}
TEST(Executor, ComposeInlineExecutor) {
InlineExecutor x;
auto xt = getKeepAliveToken(x);
ReceiverRecorder r{};
EXPECT_EQ(r.values, 0);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 0);
xt | folly::pushmi::operators::schedule() | recorder(&r) |
folly::pushmi::operators::submit();
EXPECT_EQ(r.values, 1);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 1);
}
TEST(Executor, ViaInlineExecutor) {
InlineExecutor x;
auto xt = getKeepAliveToken(x);
ReceiverRecorder r{};
EXPECT_EQ(r.values, 0);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 0);
xt | folly::pushmi::operators::schedule() | recorder(&r) |
folly::pushmi::operators::via([xt]() { return xt; }) | recorder(&r) |
folly::pushmi::operators::submit();
EXPECT_EQ(r.values, 2);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 2);
}
TEST(Executor, FlowViaOnEventBase) {
folly::EventBase evb;
folly::EventBase evb1;
std::thread t([&] { evb.loopForever(); });
std::thread t1([&] { evb1.loopForever(); });
std::array<std::string, 3> arr{{"42", "43", "44"}};
auto m = folly::pushmi::operators::flow_from(arr);
FlowReceiverRecorder r{};
EXPECT_EQ(r.values, 0);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 0);
EXPECT_EQ(r.startings, 0);
m |
folly::pushmi::operators::transform( //
[&](auto s) { return s + "a"; }) |
folly::pushmi::operators::via( //
[&]() { return getKeepAliveToken(&evb1); }) |
flow_recorder(&r) |
folly::pushmi::operators::on( //
[&]() { return getKeepAliveToken(&evb); }) |
flow_recorder(&r) |
folly::pushmi::operators::for_each( //
[](auto) {},
[&](auto) noexcept {
evb.terminateLoopSoon();
evb1.terminateLoopSoon();
},
[&]() {
evb.terminateLoopSoon();
evb1.terminateLoopSoon();
});
t.join();
t1.join();
EXPECT_EQ(r.values, 6);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 2);
EXPECT_EQ(r.startings, 2);
}
TEST(Executor, OnEventBase) {
folly::EventBase evb;
std::array<int, 3> arr{{0, 9, 99}};
auto m = folly::pushmi::operators::from(arr);
m |
folly::pushmi::operators::on(
[&]() -> Executor::KeepAlive<SequencedExecutor> {
return getKeepAliveToken(&evb);
}) |
folly::pushmi::operators::submit(
[&](auto) { EXPECT_EQ(evb.isInEventBaseThread(), true); });
evb.loop();
}
TEST(Executor, ScheduleAtEventBase) {
folly::EventBase evb(true);
auto evbt = getKeepAliveToken(evb);
ReceiverRecorder r{};
auto s =
evbt | folly::pushmi::operators::schedule_at(folly::pushmi::now(evbt));
EXPECT_EQ(r.values, 0);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 0);
std::move(s) | recorder(&r) | folly::pushmi::operators::submit();
while (r.dones == 0) {
evb.loopOnce();
}
EXPECT_EQ(r.values, 1);
EXPECT_EQ(r.errors, 0);
EXPECT_EQ(r.dones, 1);
}
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