Commit 5616bec9 authored by Dominik Charousset's avatar Dominik Charousset

Enable composable behaviors to use streaming API

parent 0ea433f3
......@@ -18,12 +18,12 @@
#pragma once
#include "caf/param.hpp"
#include "caf/abstract_composable_behavior.hpp"
#include "caf/behavior.hpp"
#include "caf/param.hpp"
#include "caf/replies_to.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_actor_pointer.hpp"
#include "caf/abstract_composable_behavior.hpp"
namespace caf {
......@@ -106,4 +106,3 @@ public:
};
} // namespace caf
......@@ -84,15 +84,27 @@ private:
flag flag_;
};
/// Convenience alias that wraps `T` into `param<T>`
/// unless `T` is arithmetic or an atom constant.
/// Converts `T` to `param<T>` unless `T` is arithmetic, an atom constant, or
/// a stream handshake.
template <class T>
using param_t =
typename std::conditional<
std::is_arithmetic<T>::value || is_atom_constant<T>::value,
T,
param<T>
>::type;
struct add_param : std::conditional<std::is_arithmetic<T>::value, T, param<T>> {
// nop
};
template <atom_value V>
struct add_param<atom_constant<V>> {
using type = atom_constant<V>;
};
template <class T>
struct add_param<stream<T>> {
using type = stream<T>;
};
/// Convenience alias that wraps `T` into `param<T>` unless `T` is arithmetic,
/// a stream handshake or an atom constant.
template <class T>
using param_t = typename add_param<T>::type;
/// Unpacks `param<T>` to `T`.
template <class T>
......@@ -112,4 +124,3 @@ struct param_decay {
};
} // namespace caf
......@@ -58,7 +58,11 @@ public:
}
/// @private
scheduled_actor* internal_ptr() const {
scheduled_actor* internal_ptr() const noexcept {
return view_.internal_ptr();
}
operator scheduled_actor*() const noexcept {
return view_.internal_ptr();
}
......
......@@ -112,23 +112,27 @@ public:
/// Returns a pointer to the sender of the current message.
/// @pre `current_mailbox_element() != nullptr`
inline strong_actor_ptr& current_sender() {
strong_actor_ptr& current_sender() {
return self_->current_sender();
}
/// Returns a pointer to the currently processed mailbox element.
inline mailbox_element* current_mailbox_element() {
mailbox_element* current_mailbox_element() {
return self_->current_mailbox_element();
}
/// @private
actor_control_block* ctrl() const {
actor_control_block* ctrl() const noexcept {
CAF_ASSERT(self_ != nullptr);
return actor_control_block::from(self_);;
}
/// @private
scheduled_actor* internal_ptr() const {
scheduled_actor* internal_ptr() const noexcept {
return self_;
}
operator scheduled_actor*() const noexcept {
return self_;
}
......@@ -137,4 +141,3 @@ private:
};
} // namespace caf
......@@ -16,27 +16,38 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE composable_behavior
#define CAF_SUITE composable_behaviors
#include "caf/test/unit_test.hpp"
#include "caf/composable_behavior.hpp"
#include "caf/all.hpp"
#include "caf/test/dsl.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
#include "caf/atom.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attach_stream_stage.hpp"
#include "caf/typed_actor.hpp"
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std;
using namespace caf;
namespace {
// -- composable behaviors using primitive data types --------------------------
// -- composable behaviors using primitive data types and streams --------------
using i3_actor = typed_actor<replies_to<int, int, int>::with<int>>;
using d_actor = typed_actor<replies_to<double>::with<double, double>>;
using source_actor = typed_actor<
replies_to<open_atom>::with<output_stream<int>>>;
using stage_actor = typed_actor<
replies_to<stream<int>>::with<output_stream<int>>>;
using sink_actor = typed_actor<reacts_to<stream<int>>>;
using foo_actor = i3_actor::extend_with<d_actor>;
class foo_actor_state : public composable_behavior<foo_actor> {
......@@ -74,12 +85,53 @@ public:
// checks whether CAF resolves "diamonds" properly by inheriting
// from two behaviors that both implement i3_actor
struct foo_actor_state2
: composed_behavior<i3_actor_state2, i3_actor_state, d_actor_state> {
: composed_behavior<i3_actor_state2, i3_actor_state, d_actor_state> {
result<int> operator()(int x, int y, int z) override {
return x - y - z;
}
};
class source_actor_state : public composable_behavior<source_actor> {
result<output_stream<int>> operator()(open_atom) override {
return attach_stream_source(
self, [](size_t& counter) { counter = 0; },
[](size_t& counter, downstream<int>& out, size_t hint) {
auto n = std::min(static_cast<size_t>(100 - counter), hint);
for (size_t i = 0; i < n; ++i)
out.push(counter++);
},
[](const int& counter) { return counter < 100; });
}
};
class stage_actor_state : public composable_behavior<stage_actor> {
result<output_stream<int>> operator()(stream<int> in) override {
return attach_stream_stage(
self, in,
[](unit_t&) {
// nop
},
[](unit_t&, downstream<int>& out, int x) {
if (x % 2 == 0)
out.push(x);
});
}
};
class sink_actor_state : public composable_behavior<sink_actor> {
std::vector<int> buf;
result<void> operator()(stream<int> in) override {
attach_stream_sink(
self, in,
[](unit_t&) {
// nop
},
[=](unit_t&, int x) { buf.emplace_back(x); });
return unit;
}
};
// -- composable behaviors using param<T> arguments ----------------------------
std::atomic<long> counting_strings_created;
......@@ -141,13 +193,13 @@ std::string to_string(const counting_string& ref) {
return ref.str();
}
} // namespace <anonymous>
} // namespace
namespace std {
template <>
struct hash<counting_string> {
inline size_t operator()(const counting_string& ref) const {
size_t operator()(const counting_string& ref) const {
hash<string> f;
return f(ref.str());
}
......@@ -163,16 +215,14 @@ using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
// "base" interface
using named_actor =
typed_actor<replies_to<get_name_atom>::with<counting_string>,
replies_to<ping_atom>::with<pong_atom>>;
using named_actor = typed_actor<
replies_to<get_name_atom>::with<counting_string>,
replies_to<ping_atom>::with<pong_atom>>;
// a simple dictionary
using dict =
named_actor::extend<replies_to<get_atom, counting_string>
::with<counting_string>,
replies_to<put_atom, counting_string, counting_string>
::with<void>>;
using dict = named_actor::extend<
replies_to<get_atom, counting_string>::with<counting_string>,
replies_to<put_atom, counting_string, counting_string>::with<void>>;
class dict_state : public composable_behavior<dict> {
public:
......@@ -204,9 +254,8 @@ protected:
std::unordered_map<counting_string, counting_string> values_;
};
using delayed_testee_actor = typed_actor<reacts_to<int>,
replies_to<bool>::with<int>,
reacts_to<std::string>>;
using delayed_testee_actor = typed_actor<
reacts_to<int>, replies_to<bool>::with<int>, reacts_to<std::string>>;
class delayed_testee : public composable_behavior<delayed_testee_actor> {
public:
......@@ -228,142 +277,135 @@ public:
}
};
struct fixture {
fixture() : system(cfg) {
// nop
struct config : actor_system_config {
config() {
using foo_actor_impl = composable_behavior_based_actor<foo_actor_state>;
add_actor_type<foo_actor_impl>("foo_actor");
}
};
actor_system_config cfg;
actor_system system;
struct fixture : test_coordinator_fixture<config> {
// nop
};
} // namespace <anonymous>
} // namespace
CAF_TEST_FIXTURE_SCOPE(composable_behaviors_tests, fixture)
CAF_TEST(composition) {
// test foo_foo_actor_state
auto f1 = make_function_view(system.spawn<foo_actor_state>());
CAF_CHECK_EQUAL(f1(1, 2, 4), 7);
CAF_CHECK_EQUAL(f1(42.0), std::make_tuple(42.0, 42.0));
// test on-the-fly composition of i3_actor_state and d_actor_state
f1.assign(system.spawn<composed_behavior<i3_actor_state, d_actor_state>>());
CAF_CHECK_EQUAL(f1(1, 2, 4), 7);
CAF_CHECK_EQUAL(f1(42.0), std::make_tuple(42.0, 42.0));
// test on-the-fly composition of i3_actor_state2 and d_actor_state
f1.assign(system.spawn<composed_behavior<i3_actor_state2, d_actor_state>>());
CAF_CHECK_EQUAL(f1(1, 2, 4), 8);
CAF_CHECK_EQUAL(f1(42.0), std::make_tuple(42.0, 42.0));
// test foo_actor_state2
f1.assign(system.spawn<foo_actor_state2>());
CAF_CHECK_EQUAL(f1(1, 2, 4), -5);
CAF_CHECK_EQUAL(f1(42.0), std::make_tuple(42.0, 42.0));
CAF_MESSAGE("test foo_actor_state");
auto f1 = sys.spawn<foo_actor_state>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(7));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test composed_behavior<i3_actor_state, d_actor_state>");
f1 = sys.spawn<composed_behavior<i3_actor_state, d_actor_state>>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(7));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test composed_behavior<i3_actor_state2, d_actor_state>");
f1 = sys.spawn<composed_behavior<i3_actor_state2, d_actor_state>>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(8));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
CAF_MESSAGE("test foo_actor_state2");
f1 = sys.spawn<foo_actor_state2>();
inject((int, int, int), from(self).to(f1).with(1, 2, 4));
expect((int), from(f1).to(self).with(-5));
inject((double), from(self).to(f1).with(42.0));
expect((double, double), from(f1).to(self).with(42.0, 42.0));
}
CAF_TEST(param_detaching) {
auto dict = actor_cast<actor>(system.spawn<dict_state>());
scoped_actor self{system};
// this ping-pong makes sure that dict has cleaned up all state related
// to a test before moving to the second test; otherwise, reference counts
// can diverge from what we expect
auto ping_pong = [&] {
self->request(dict, infinite, ping_atom::value).receive(
[](pong_atom) {
// nop
},
[&](error& err) {
CAF_FAIL("error: " << system.render(err));
}
);
};
auto dict = actor_cast<actor>(sys.spawn<dict_state>());
// Using CAF is the key to success!
counting_string key = "CAF";
counting_string value = "success";
CAF_CHECK_EQUAL(counting_strings_created.load(), 2);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 0);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 0);
// wrap two strings into messages
// Wrap two strings into messages.
auto put_msg = make_message(put_atom::value, key, value);
auto get_msg = make_message(get_atom::value, key);
CAF_CHECK_EQUAL(counting_strings_created.load(), 5);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 0);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 0);
// send put message to dictionary
self->request(dict, infinite, put_msg).receive(
[&] {
ping_pong();
// the handler of put_atom calls .move() on key and value,
// both causing to detach + move into the map
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
},
ERROR_HANDLER
);
// send put message to dictionary again
self->request(dict, infinite, put_msg).receive(
[&] {
ping_pong();
// the handler checks whether key already exists -> no copies
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
},
ERROR_HANDLER
);
// alter our initial put, this time moving it to the dictionary
// Send put message to dictionary.
self->send(dict, put_msg);
sched.run();
// The handler of put_atom calls .move() on key and value, both causing to
// detach + move into the map.
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
// Send put message to dictionary again.
self->send(dict, put_msg);
sched.run();
// The handler checks whether key already exists -> no copies.
CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
// Alter our initial put, this time moving it to the dictionary.
put_msg.get_mutable_as<counting_string>(1) = "neverlord";
put_msg.get_mutable_as<counting_string>(2) = "CAF";
// send put message to dictionary
self->request(dict, infinite, std::move(put_msg)).receive(
[&] {
ping_pong();
// the handler of put_atom calls .move() on key and value,
// but no detaching occurs this time (unique access) -> move into the map
CAF_CHECK_EQUAL(counting_strings_created.load(), 11);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 4);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 4);
},
ERROR_HANDLER
);
// finally, check for original key
self->request(dict, infinite, std::move(get_msg)).receive(
[&](const counting_string& str) {
ping_pong();
// we receive a copy of the value, which is copied out of the map and
// then moved into the result message;
// the string from our get_msg is destroyed
CAF_CHECK_EQUAL(counting_strings_created.load(), 13);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 5);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 6);
CAF_CHECK_EQUAL(str, "success");
},
ERROR_HANDLER
);
// temporary of our handler is destroyed
// Send new put message to dictionary.
self->send(dict, std::move(put_msg));
sched.run();
// The handler of put_atom calls .move() on key and value, but no detaching
// occurs this time (unique access) -> move into the map.
CAF_CHECK_EQUAL(counting_strings_created.load(), 11);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 4);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 4);
// Finally, check for original key.
self->send(dict, std::move(get_msg));
sched.run();
self->receive([&](const counting_string& str) {
// We receive a copy of the value, which is copied out of the map and then
// moved into the result message; the string from our get_msg is destroyed.
CAF_CHECK_EQUAL(counting_strings_created.load(), 13);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 5);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 6);
CAF_CHECK_EQUAL(str, "success");
});
// Temporary of our handler is destroyed.
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 7);
self->send_exit(dict, exit_reason::kill);
self->await_all_other_actors_done();
// only `key` and `value` from this scope remain
self->send_exit(dict, exit_reason::user_shutdown);
sched.run();
dict = nullptr;
// Only `key` and `value` from this scope remain.
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 11);
}
CAF_TEST(delayed_sends) {
scoped_actor self{system};
auto testee = self->spawn<delayed_testee>();
self->send(testee, 42);
inject((int), from(self).to(testee).with(42));
disallow((bool), from(_).to(testee));
sched.trigger_timeouts();
expect((bool), from(_).to(testee));
disallow((std::string), from(testee).to(testee).with("hello"));
sched.trigger_timeouts();
expect((std::string), from(testee).to(testee).with("hello"));
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST(dynamic_spawning) {
using impl = composable_behavior_based_actor<foo_actor_state>;
actor_system_config cfg;
cfg.add_actor_type<impl>("foo_actor");
actor_system sys{cfg};
auto sr = sys.spawn<foo_actor>("foo_actor", make_message());
CAF_REQUIRE(sr);
auto f1 = make_function_view(std::move(*sr));
CAF_CHECK_EQUAL(f1(1, 2, 4), 7);
CAF_CHECK_EQUAL(f1(42.0), std::make_tuple(42.0, 42.0));
auto testee = unbox(sys.spawn<foo_actor>("foo_actor", make_message()));
inject((int, int, int), from(self).to(testee).with(1, 2, 4));
expect((int), from(testee).to(self).with(7));
inject((double), from(self).to(testee).with(42.0));
expect((double, double), from(testee).to(self).with(42.0, 42.0));
}
CAF_TEST(streaming) {
// TODO: currently broken
// auto src = sys.spawn<source_actor_state>();
// auto stg = sys.spawn<stage_actor_state>();
// auto snk = sys.spawn<sink_actor_state>();
// auto pipeline = snk * stg * src;
// self->send(pipeline, open_atom::value);
// sched.run();
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -18,13 +18,14 @@
#pragma once
#include <type_traits>
#include "caf/all.hpp"
#include "caf/config.hpp"
#include "caf/detail/gcd.hpp"
#include "caf/meta/annotation.hpp"
#include "caf/test/unit_test.hpp"
#include "caf/detail/gcd.hpp"
CAF_PUSH_WARNINGS
/// The type of `_`.
......@@ -375,6 +376,71 @@ protected:
std::function<void ()> peek_;
};
template <class... Ts>
class inject_clause {
public:
inject_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), dest_(nullptr) {
// nop
}
inject_clause(inject_clause&& other) = default;
template <class Handle>
inject_clause& from(const Handle& whom) {
src_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return *this;
}
template <class Handle>
inject_clause& to(const Handle& whom) {
dest_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return *this;
}
inject_clause& to(const caf::scoped_actor& whom) {
dest_ = whom.ptr();
return *this;
}
void with(Ts... xs) {
if (src_ == nullptr)
CAF_FAIL("missing .from() in inject() statement");
if (dest_ == nullptr)
CAF_FAIL("missing .to() in inject() statement");
caf::send_as(caf::actor_cast<caf::actor>(src_),
caf::actor_cast<caf::actor>(dest_), xs...);
CAF_REQUIRE(sched_.prioritize(dest_));
auto dest_ptr = &sched_.next_job<caf::abstract_actor>();
auto ptr = dest_ptr->peek_at_next_mailbox_element();
CAF_REQUIRE(ptr != nullptr);
CAF_REQUIRE_EQUAL(ptr->sender, src_);
// TODO: replace this workaround with the make_tuple() line when dropping
// support for GCC 4.8.
std::tuple<Ts...> tmp{std::move(xs)...};
using namespace caf::detail;
elementwise_compare_inspector<decltype(tmp)> inspector{tmp};
auto ys = extract<Ts...>(dest_);
auto ys_indices = get_indices(ys);
CAF_REQUIRE(apply_args(inspector, ys_indices, ys));
run_once();
}
protected:
void run_once() {
auto ptr = caf::actor_cast<caf::abstract_actor*>(dest_);
auto dptr = dynamic_cast<caf::blocking_actor*>(ptr);
if (dptr == nullptr)
sched_.run_once();
else
dptr->dequeue(); // Drop message.
}
caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_;
caf::strong_actor_ptr dest_;
};
template <class... Ts>
class allow_clause {
public:
......@@ -796,6 +862,12 @@ T unbox(T* x) {
expect_clause<CAF_EXPAND(CAF_DSL_LIST types)>{sched}.fields; \
} while (false)
#define inject(types, fields) \
do { \
CAF_MESSAGE("inject" << #types << "." << #fields); \
inject_clause<CAF_EXPAND(CAF_DSL_LIST types)>{sched}.fields; \
} while (false)
/// Convenience macro for defining allow clauses.
#define allow(types, fields) \
([&] { \
......
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