Commit 09cd843a authored by Dominik Charousset's avatar Dominik Charousset

Implement "local" side for actor migration

parent 36a611a3
...@@ -171,6 +171,8 @@ public: ...@@ -171,6 +171,8 @@ public:
static constexpr int is_blocking_flag = 0x10; // blocking_actor static constexpr int is_blocking_flag = 0x10; // blocking_actor
static constexpr int is_detached_flag = 0x20; // local_actor static constexpr int is_detached_flag = 0x20; // local_actor
static constexpr int is_priority_aware_flag = 0x40; // local_actor static constexpr int is_priority_aware_flag = 0x40; // local_actor
static constexpr int is_serializable_flag = 0x40; // local_actor
static constexpr int is_migrated_from_flag = 0x80; // local_actor
inline void set_flag(bool enable_flag, int mask) { inline void set_flag(bool enable_flag, int mask) {
auto x = flags(); auto x = flags();
...@@ -227,6 +229,22 @@ public: ...@@ -227,6 +229,22 @@ public:
set_flag(value, is_priority_aware_flag); set_flag(value, is_priority_aware_flag);
} }
inline bool is_serializable() const {
return get_flag(is_serializable_flag);
}
inline void is_serializable(bool value) {
set_flag(value, is_serializable_flag);
}
inline bool is_migrated_from() const {
return get_flag(is_migrated_from_flag);
}
inline void is_migrated_from(bool value) {
set_flag(value, is_migrated_from_flag);
}
// Tries to run a custom exception handler for `eptr`. // Tries to run a custom exception handler for `eptr`.
optional<uint32_t> handle(const std::exception_ptr& eptr); optional<uint32_t> handle(const std::exception_ptr& eptr);
......
...@@ -20,9 +20,9 @@ ...@@ -20,9 +20,9 @@
#ifndef CAF_ACTOR_NAMESPACE_HPP #ifndef CAF_ACTOR_NAMESPACE_HPP
#define CAF_ACTOR_NAMESPACE_HPP #define CAF_ACTOR_NAMESPACE_HPP
#include <map>
#include <utility> #include <utility>
#include <functional> #include <functional>
#include <unordered_map>
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
...@@ -56,7 +56,7 @@ public: ...@@ -56,7 +56,7 @@ public:
/// Writes an actor address to `sink` and adds the actor /// Writes an actor address to `sink` and adds the actor
/// to the list of known actors for a later deserialization. /// to the list of known actors for a later deserialization.
void write(serializer* sink, const actor_addr& ptr); void write(serializer* sink, const actor_addr& ptr) const;
/// Reads an actor address from `source,` creating /// Reads an actor address from `source,` creating
/// addresses for remote actors on the fly if needed. /// addresses for remote actors on the fly if needed.
...@@ -120,7 +120,7 @@ public: ...@@ -120,7 +120,7 @@ public:
private: private:
backend& backend_; backend& backend_;
std::map<key_type, proxy_map> proxies_; std::unordered_map<key_type, proxy_map> proxies_;
}; };
} // namespace caf } // namespace caf
......
...@@ -123,6 +123,9 @@ using close_atom = atom_constant<atom("CLOSE")>; ...@@ -123,6 +123,9 @@ using close_atom = atom_constant<atom("CLOSE")>;
/// Generic 'SPAWN' atom, e.g., for spawning remote actors. /// Generic 'SPAWN' atom, e.g., for spawning remote actors.
using spawn_atom = atom_constant<atom("SPAWN")>; using spawn_atom = atom_constant<atom("SPAWN")>;
/// Atom to signalize an actor to migrate its state to another actor.
using migrate_atom = atom_constant<atom("MIGRATE")>;
} // namespace caf } // namespace caf
namespace std { namespace std {
......
...@@ -127,6 +127,11 @@ operator>>(deserializer& source, T& value) { ...@@ -127,6 +127,11 @@ operator>>(deserializer& source, T& value) {
return source.read(value, uniform_typeid<T>()); return source.read(value, uniform_typeid<T>());
} }
template <class T>
void operator&(deserializer& source, T& value) {
source >> value;
}
} // namespace caf } // namespace caf
#endif // CAF_DESERIALIZER_HPP #endif // CAF_DESERIALIZER_HPP
...@@ -493,6 +493,26 @@ public: ...@@ -493,6 +493,26 @@ public:
static constexpr bool value = false; static constexpr bool value = false;
}; };
// checks whether T is serializable using a free function `serialize` taking
// either a `caf::serializer` or `caf::deserializer` as first argument
template <class T>
struct is_serializable {
private:
template <class U>
static int8_t fun(U*, decltype(serialize(std::declval<serializer&>(),
std::declval<U&>(), 0))* = nullptr,
decltype(serialize(std::declval<deserializer&>(),
std::declval<U&>(), 0))* = nullptr);
static int16_t fun(...);
public:
static constexpr bool value = sizeof(fun(static_cast<T*>(nullptr))) == 1;
};
template <class T>
constexpr bool is_serializable<T>::value;
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -29,6 +29,18 @@ ...@@ -29,6 +29,18 @@
namespace caf { namespace caf {
namespace experimental { namespace experimental {
template <class Archive, class U>
typename std::enable_if<detail::is_serializable<U>::value>::type
serialize_state(Archive& ar, U& st, const unsigned int version) {
serialize(ar, st, version);
}
template <class Archive, class U>
typename std::enable_if<! detail::is_serializable<U>::value>::type
serialize_state(Archive&, U&, const unsigned int) {
throw std::logic_error("serialize_state with unserializable type called");
}
/// An event-based actor with managed state. The state is constructed /// An event-based actor with managed state. The state is constructed
/// before `make_behavior` will get called and destroyed after the /// before `make_behavior` will get called and destroyed after the
/// actor called `quit`. This state management brakes cycles and /// actor called `quit`. This state management brakes cycles and
...@@ -39,7 +51,8 @@ class stateful_actor : public Base { ...@@ -39,7 +51,8 @@ class stateful_actor : public Base {
public: public:
template <class... Ts> template <class... Ts>
stateful_actor(Ts&&... xs) : Base(std::forward<Ts>(xs)...), state(state_) { stateful_actor(Ts&&... xs) : Base(std::forward<Ts>(xs)...), state(state_) {
// nop if (detail::is_serializable<State>::value)
this->is_serializable(true);
} }
~stateful_actor() { ~stateful_actor() {
...@@ -55,6 +68,14 @@ public: ...@@ -55,6 +68,14 @@ public:
return get_name(state_); return get_name(state_);
} }
void save(serializer& sink, const unsigned int version) override {
serialize_state(sink, state, version);
}
void load(deserializer& source, const unsigned int version) override {
serialize_state(source, state, version);
}
/// A reference to the actor's state. /// A reference to the actor's state.
State& state; State& state;
......
...@@ -392,6 +392,16 @@ public: ...@@ -392,6 +392,16 @@ public:
/// implementation simply returns "actor". /// implementation simply returns "actor".
virtual const char* name() const; virtual const char* name() const;
/// Serializes the state of this actor to `sink`. This function is
/// only called if this actor has set the `is_serializable` flag.
/// The default implementation throws a `std::logic_error`.
virtual void save(serializer& sink, const unsigned int version);
/// Deserializes the state of this actor from `source`. This function is
/// only called if this actor has set the `is_serializable` flag.
/// The default implementation throws a `std::logic_error`.
virtual void load(deserializer& source, const unsigned int version);
/**************************************************************************** /****************************************************************************
* deprecated member functions * * deprecated member functions *
****************************************************************************/ ****************************************************************************/
...@@ -613,9 +623,9 @@ public: ...@@ -613,9 +623,9 @@ public:
initial_behavior_fac_ = std::move(fun); initial_behavior_fac_ = std::move(fun);
} }
protected:
void do_become(behavior bhvr, bool discard_old); void do_become(behavior bhvr, bool discard_old);
protected:
// used only in thread-mapped actors // used only in thread-mapped actors
void await_data(); void await_data();
......
...@@ -107,6 +107,11 @@ operator<<(serializer& sink, const T& value) { ...@@ -107,6 +107,11 @@ operator<<(serializer& sink, const T& value) {
return sink.write(value, uniform_typeid<T>()); return sink.write(value, uniform_typeid<T>());
} }
template <class T>
void operator&(serializer& sink, const T& value) {
sink << value;
}
} // namespace caf } // namespace caf
#endif // CAF_SERIALIZER_HPP #endif // CAF_SERIALIZER_HPP
...@@ -62,7 +62,7 @@ actor_namespace::actor_namespace(backend& be) : backend_(be) { ...@@ -62,7 +62,7 @@ actor_namespace::actor_namespace(backend& be) : backend_(be) {
// nop // nop
} }
void actor_namespace::write(serializer* sink, const actor_addr& addr) { void actor_namespace::write(serializer* sink, const actor_addr& addr) const {
CAF_ASSERT(sink != nullptr); CAF_ASSERT(sink != nullptr);
if (! addr) { if (! addr) {
node_id::host_id_type zero; node_id::host_id_type zero;
......
...@@ -186,27 +186,100 @@ enum class msg_type { ...@@ -186,27 +186,100 @@ enum class msg_type {
expired_sync_response, // a sync response that already timed out expired_sync_response, // a sync response that already timed out
timeout, // triggers currently active timeout timeout, // triggers currently active timeout
ordinary, // an asynchronous message or sync. request ordinary, // an asynchronous message or sync. request
sync_response // a synchronous response sync_response, // a synchronous response
sys_message // a system message, e.g., signalizing migration
}; };
msg_type filter_msg(local_actor* self, mailbox_element& node) { msg_type filter_msg(local_actor* self, mailbox_element& node) {
const message& msg = node.msg; message& msg = node.msg;
auto mid = node.mid; auto mid = node.mid;
if (mid.is_response()) { if (mid.is_response())
return self->awaits(mid) ? msg_type::sync_response return self->awaits(mid) ? msg_type::sync_response
: msg_type::expired_sync_response; : msg_type::expired_sync_response;
// intercept system messages, e.g., signalizing migration
if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) {
bool mismatch = false;
msg.apply({
[&](sys_atom, migrate_atom, const actor& mm) {
// migrate this actor to `target`
if (! self->is_serializable()) {
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
error_atom::value, "not serializable"),
self->host());
return;
}
std::vector<char> buf;
binary_serializer bs{std::back_inserter(buf)};
self->save(bs, 0);
auto sender = node.sender;
auto mid = node.mid;
// sync_send(...)
auto req = self->sync_send_impl(message_priority::normal, mm,
migrate_atom::value, self->name(),
std::move(buf));
self->set_response_handler(req, behavior{
[=](ok_atom, const actor_addr& dest) {
// respond to original message with {'OK', dest}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(),
ok_atom::value, dest),
self->host());
// "decay" into a proxy for `dest`
auto dest_hdl = actor_cast<actor>(dest);
self->do_become(behavior{
others >> [=] {
self->forward_current_message(dest_hdl);
}
}, false);
self->is_migrated_from(true);
},
[=](error_atom, std::string& errmsg) {
// respond to original message with {'ERROR', errmsg}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(),
error_atom::value,
std::move(errmsg)),
self->host());
}
});
},
[&](sys_atom, migrate_atom, std::vector<char>& buf) {
// "replace" this actor with the content of `buf`
if (! self->is_serializable()) {
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
error_atom::value, "not serializable"),
self->host());
return;
}
if (self->is_migrated_from()) {
// undo the `do_become` we did when migrating away from this object
self->bhvr_stack().pop_back();
self->is_migrated_from(false);
}
binary_deserializer bd{buf.data(), buf.size()};
self->load(bd, 0);
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
ok_atom::value, self->address()),
self->host());
},
others >> [&] {
mismatch = true;
}
});
return mismatch ? msg_type::ordinary : msg_type::sys_message;
} }
if (msg.size() != 1) { // all other system messages always consist of one element
if (msg.size() != 1)
return msg_type::ordinary; return msg_type::ordinary;
}
if (msg.match_element<timeout_msg>(0)) { if (msg.match_element<timeout_msg>(0)) {
auto& tm = msg.get_as<timeout_msg>(0); auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id; auto tid = tm.timeout_id;
CAF_ASSERT(! mid.valid()); CAF_ASSERT(! mid.valid());
if (self->is_active_timeout(tid)) { return self->is_active_timeout(tid) ? msg_type::timeout
return msg_type::timeout; : msg_type::expired_timeout;
}
return msg_type::expired_timeout;
} }
if (msg.match_element<exit_msg>(0)) { if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0); auto& em = msg.get_as<exit_msg>(0);
...@@ -322,6 +395,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -322,6 +395,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
case msg_type::expired_timeout: case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message"); CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped; return im_dropped;
case msg_type::sys_message:
CAF_LOG_DEBUG("handled system message");
return im_dropped;
case msg_type::non_normal_exit: case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal"); CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled // this message was handled
...@@ -829,6 +905,14 @@ const char* local_actor::name() const { ...@@ -829,6 +905,14 @@ const char* local_actor::name() const {
return "actor"; return "actor";
} }
void local_actor::save(serializer&, const unsigned int) {
throw std::logic_error("local_actor::serialize called");
}
void local_actor::load(deserializer&, const unsigned int) {
throw std::logic_error("local_actor::deserialize called");
}
behavior& local_actor::get_behavior() { behavior& local_actor::get_behavior() {
return pending_responses_.empty() ? bhvr_stack_.back() return pending_responses_.empty() ? bhvr_stack_.back()
: pending_responses_.front().second; : pending_responses_.front().second;
......
...@@ -95,18 +95,6 @@ const char* numbered_type_names[] = { ...@@ -95,18 +95,6 @@ const char* numbered_type_names[] = {
namespace { namespace {
// might become part of serializer's public API at some point
template <class T>
void operator&(serializer& sink, const T& value) {
sink << value;
}
// might become part of deserializer's public API at some point
template <class T>
void operator&(deserializer& source, T& value) {
source >> value;
}
// primitive types are handled by serializer/deserializer directly // primitive types are handled by serializer/deserializer directly
template <class T, class U> template <class T, class U>
typename std::enable_if< typename std::enable_if<
......
...@@ -69,7 +69,7 @@ struct fixture { ...@@ -69,7 +69,7 @@ struct fixture {
} }
~fixture() { ~fixture() {
{ // lifetime scope of scoped_actor if (aut != invalid_actor) {
scoped_actor self; scoped_actor self;
self->monitor(aut); self->monitor(aut);
self->receive( self->receive(
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE announce_actor_type
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#include "caf/detail/actor_registry.hpp"
using namespace caf;
using namespace caf::experimental;
using std::endl;
namespace {
struct fixture {
actor aut;
actor spawner;
fixture() {
auto registry = detail::singletons::get_actor_registry();
spawner = registry->get_named(atom("spawner"));
}
void set_aut(message args, bool expect_fail = false) {
CAF_MESSAGE("set aut");
scoped_actor self;
self->on_sync_failure([&] {
CAF_TEST_ERROR("received unexpeced sync. response: "
<< to_string(self->current_message()));
});
if (expect_fail) {
self->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](error_atom, const std::string&) {
CAF_TEST_VERBOSE("received error_atom (expected)");
}
);
} else {
self->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](ok_atom, actor_addr res, const std::set<std::string>& ifs) {
CAF_REQUIRE(res != invalid_actor_addr);
aut = actor_cast<actor>(res);
CAF_CHECK(ifs.empty());
}
);
}
}
~fixture() {
if (aut != invalid_actor) {
scoped_actor self;
self->monitor(aut);
self->receive(
[](const down_msg& dm) {
CAF_CHECK(dm.reason == exit_reason::normal);
}
);
}
await_all_actors_done();
shutdown();
}
};
using detail::is_serializable;
// not serializable
class testee1 {
};
// serializable via member function
class testee2 {
public:
template <class Archive>
void serialize(Archive&, const unsigned int) {
// nop
}
};
// serializable via free function
class testee3 {
// nop
};
template<class Archive>
void serialize(Archive&, testee3&, const unsigned int) {
// nop
}
template <class Archive, class T>
void serialize(Archive& ar, T& x, const unsigned int version,
decltype(x.serialize(ar, version))* = nullptr) {
x.serialize(ar, version);
}
struct migratable_state {
int value = 0;
};
template <class Archive>
void serialize(Archive& ar, migratable_state& x, const unsigned int) {
ar & x.value;
}
struct migratable_actor : stateful_actor<migratable_state> {
behavior make_behavior() override {
return {
[=](get_atom) {
return state.value;
},
[=](put_atom, int value) {
state.value = value;
},
[=](sys_atom, migrate_atom, actor_addr destination) {
std::vector<char> buf;
binary_serializer bs{std::back_inserter(buf)};
save(bs, 0);
actor dest = actor_cast<actor>(destination);
link_to(dest);
sync_send(dest, sys_atom::value, migrate_atom::value, std::move(buf)).then(
[=](ok_atom) {
// "decay" into a proxy for `dest`
become(
[=](sys_atom, migrate_atom, std::vector<char> buf) {
unlink_from(dest);
become(make_behavior());
binary_deserializer bd{buf.data(), buf.size()};
deserialize(bd, 0);
return make_message(ok_atom::value);
},
others >> [=] {
forward_to(dest);
}
);
},
others >> [=] {
// do nothing, i.e., process further messages as usual
}
);
},
[=](sys_atom, migrate_atom, std::vector<char> buf) {
binary_deserializer bd{buf.data(), buf.size()};
deserialize(bd, 0);
return make_message(ok_atom::value);
}
};
}
};
} // namespace <anonymous>
CAF_TEST(serializable_stuff) {
CAF_CHECK(is_serializable<testee1>::value == false);
CAF_CHECK(is_serializable<testee2>::value == true);
CAF_CHECK(is_serializable<testee3>::value == true);
}
CAF_TEST(migratable_stuff) {
auto a = spawn<migratable_actor>();
auto b = spawn<migratable_actor>();
scoped_actor self;
self->send(a, put_atom::value, 42);
self->send(a, sys_atom::value, migrate_atom::value, b.address());
self->sync_send(a, get_atom::value).await(
[&](int result) {
CAF_CHECK(result == 42);
CAF_CHECK(self->current_sender() == b.address());
}
);
self->send_exit(b, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST_FIXTURE_SCOPE(announce_actor_type_tests, fixture)
CAF_TEST(fun_no_args) {
auto test_actor = [] {
CAF_MESSAGE("inside test_actor");
};
announce_actor_type("test_actor", test_actor);
set_aut(make_message());
}
CAF_TEST(fun_no_args_selfptr) {
auto test_actor = [](event_based_actor*) {
CAF_MESSAGE("inside test_actor");
};
announce_actor_type("test_actor", test_actor);
set_aut(make_message());
}
CAF_TEST(fun_one_arg) {
auto test_actor = [](int i) {
CAF_CHECK_EQUAL(i, 42);
};
announce_actor_type("test_actor", test_actor);
set_aut(make_message(42));
}
CAF_TEST(fun_one_arg_selfptr) {
auto test_actor = [](event_based_actor*, int i) {
CAF_CHECK_EQUAL(i, 42);
};
announce_actor_type("test_actor", test_actor);
set_aut(make_message(42));
}
CAF_TEST(class_no_arg) {
struct test_actor : event_based_actor {
behavior make_behavior() override {
return {};
}
};
announce_actor_type<test_actor>("test_actor");
set_aut(make_message(42), true);
set_aut(make_message());
}
CAF_TEST(class_one_arg) {
struct test_actor : event_based_actor {
test_actor(int value) {
CAF_CHECK_EQUAL(value, 42);
}
behavior make_behavior() override {
return {};
}
};
announce_actor_type<test_actor, const int&>("test_actor");
set_aut(make_message(), true);
set_aut(make_message(42));
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE local_migration
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
#include "caf/detail/actor_registry.hpp"
using namespace caf;
using namespace caf::experimental;
using std::endl;
namespace {
struct migratable_state {
int value = 0;
static const char* name;
};
const char* migratable_state::name = "migratable_actor";
template <class Archive>
void serialize(Archive& ar, migratable_state& x, const unsigned int) {
ar & x.value;
}
struct migratable_actor : stateful_actor<migratable_state> {
behavior make_behavior() override {
return {
[=](get_atom) {
return state.value;
},
[=](put_atom, int value) {
state.value = value;
}
};
}
};
// always migrates to `dest`
behavior pseudo_mm(event_based_actor* self, const actor& dest) {
return {
[=](migrate_atom, const std::string& name, std::vector<char>& buf) {
CAF_CHECK(name == "migratable_actor");
self->delegate(dest, sys_atom::value, migrate_atom::value,
std::move(buf));
}
};
}
} // namespace <anonymous>
CAF_TEST(migrate_locally) {
auto a = spawn<migratable_actor>();
auto b = spawn<migratable_actor>();
auto mm1 = spawn(pseudo_mm, b);
scoped_actor self;
self->send(a, put_atom::value, 42);
// migrate from a to b
self->sync_send(a, sys_atom::value, migrate_atom::value, mm1).await(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == b);
}
);
self->sync_send(a, get_atom::value).await(
[&](int result) {
CAF_CHECK(result == 42);
CAF_CHECK(self->current_sender() == b.address());
}
);
auto mm2 = spawn(pseudo_mm, a);
self->send(b, put_atom::value, 23);
// migrate back from b to a
self->sync_send(b, sys_atom::value, migrate_atom::value, mm2).await(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == a);
}
);
self->sync_send(b, get_atom::value).await(
[&](int result) {
CAF_CHECK(result == 23);
CAF_CHECK(self->current_sender() == a.address());
}
);
self->send_exit(a, exit_reason::kill);
self->send_exit(b, exit_reason::kill);
self->send_exit(mm1, exit_reason::kill);
self->send_exit(mm2, exit_reason::kill);
self->await_all_other_actors_done();
}
...@@ -448,9 +448,9 @@ CAF_TEST(remote_actor_and_send) { ...@@ -448,9 +448,9 @@ CAF_TEST(remote_actor_and_send) {
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
mpx()->provide_scribe("localhost", 4242, remote_hdl(0)); mpx()->provide_scribe("localhost", 4242, remote_hdl(0));
CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 1); CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 1);
auto mm = get_middleman_actor(); auto mm1 = get_middleman_actor();
actor result; actor result;
auto f = self()->sync_send(mm, connect_atom::value, auto f = self()->sync_send(mm1, connect_atom::value,
"localhost", uint16_t{4242}); "localhost", uint16_t{4242});
mpx()->exec_runnable(); // process message in basp_broker mpx()->exec_runnable(); // process message in basp_broker
CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 0); CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 0);
......
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