Commit b45f558f authored by Dominik Charousset's avatar Dominik Charousset

Add `announce_actor_type`, revise middleman actor

parent 456e7d04
...@@ -35,7 +35,6 @@ using plus_atom = atom_constant<atom("plus")>; ...@@ -35,7 +35,6 @@ using plus_atom = atom_constant<atom("plus")>;
using minus_atom = atom_constant<atom("minus")>; using minus_atom = atom_constant<atom("minus")>;
using result_atom = atom_constant<atom("result")>; using result_atom = atom_constant<atom("result")>;
using rebind_atom = atom_constant<atom("rebind")>; using rebind_atom = atom_constant<atom("rebind")>;
using connect_atom = atom_constant<atom("connect")>;
using reconnect_atom = atom_constant<atom("reconnect")>; using reconnect_atom = atom_constant<atom("reconnect")>;
// our "service" // our "service"
...@@ -105,9 +104,13 @@ private: ...@@ -105,9 +104,13 @@ private:
behavior reconnecting(std::function<void()> continuation = nullptr) { behavior reconnecting(std::function<void()> continuation = nullptr) {
using std::chrono::seconds; using std::chrono::seconds;
auto mm = io::get_middleman_actor(); auto mm = io::get_middleman_actor();
send(mm, get_atom::value, host_, port_); send(mm, connect_atom::value, host_, port_);
return { return {
[=](ok_atom, const actor_addr& new_server) { [=](ok_atom, node_id&, actor_addr& new_server, std::set<std::string>&) {
if (new_server == invalid_actor_addr) {
aout(this) << "*** received invalid remote actor" << endl;
return;
}
aout(this) << "*** connection succeeded, awaiting tasks" << endl; aout(this) << "*** connection succeeded, awaiting tasks" << endl;
server_ = actor_cast<actor>(new_server); server_ = actor_cast<actor>(new_server);
// return to previous behavior // return to previous behavior
...@@ -122,7 +125,7 @@ private: ...@@ -122,7 +125,7 @@ private:
<< ": " << errstr << ": " << errstr
<< " [try again in 3s]" << " [try again in 3s]"
<< endl; << endl;
delayed_send(mm, seconds(3), get_atom::value, host_, port_); delayed_send(mm, seconds(3), connect_atom::value, host_, port_);
}, },
[=](rebind_atom, string& nhost, uint16_t nport) { [=](rebind_atom, string& nhost, uint16_t nport) {
aout(this) << "*** rebind to " << nhost << ":" << nport << endl; aout(this) << "*** rebind to " << nhost << ":" << nport << endl;
...@@ -131,13 +134,17 @@ private: ...@@ -131,13 +134,17 @@ private:
swap(port_, nport); swap(port_, nport);
auto send_mm = [=] { auto send_mm = [=] {
unbecome(); unbecome();
send(mm, get_atom::value, host_, port_); send(mm, connect_atom::value, host_, port_);
}; };
// await pending ok/error message first, then send new request to MM // await pending ok/error message first, then send new request to MM
become( become(
keep_behavior, keep_behavior,
[=](ok_atom&, actor_addr&) { send_mm(); }, [=](ok_atom&, actor_addr&) {
[=](error_atom&, string&) { send_mm(); } send_mm();
},
[=](error_atom&, string&) {
send_mm();
}
); );
}, },
// simply ignore all requests until we have a connection // simply ignore all requests until we have a connection
......
...@@ -24,6 +24,7 @@ set (LIBCAF_CORE_SRCS ...@@ -24,6 +24,7 @@ set (LIBCAF_CORE_SRCS
src/actor_proxy.cpp src/actor_proxy.cpp
src/actor_registry.cpp src/actor_registry.cpp
src/attachable.cpp src/attachable.cpp
src/announce_actor_type.cpp
src/behavior.cpp src/behavior.cpp
src/behavior_stack.cpp src/behavior_stack.cpp
src/behavior_impl.cpp src/behavior_impl.cpp
......
...@@ -135,6 +135,11 @@ public: ...@@ -135,6 +135,11 @@ public:
/// Exchange content of `*this` and `other`. /// Exchange content of `*this` and `other`.
void swap(actor& other) noexcept; void swap(actor& other) noexcept;
/// Returns the interface definition for this actor handle.
static std::set<std::string> message_types() {
return std::set<std::string>{};
}
/// @cond PRIVATE /// @cond PRIVATE
inline abstract_actor* operator->() const noexcept { inline abstract_actor* operator->() const noexcept {
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_ATOM_HPP #define CAF_ATOM_HPP
#include <string> #include <string>
#include <functional>
#include <type_traits> #include <type_traits>
#include "caf/detail/atom_val.hpp" #include "caf/detail/atom_val.hpp"
...@@ -104,6 +105,33 @@ using link_atom = atom_constant<atom("LINK")>; ...@@ -104,6 +105,33 @@ using link_atom = atom_constant<atom("LINK")>;
/// Generic 'UNLINK' atom for removing networked links. /// Generic 'UNLINK' atom for removing networked links.
using unlink_atom = atom_constant<atom("UNLINK")>; using unlink_atom = atom_constant<atom("UNLINK")>;
/// Generic 'PUBLISH' atom, e.g., for publishing actors at a given port.
using publish_atom = atom_constant<atom("PUBLISH")>;
/// Generic 'UNPUBLISH' atom, e.g., for removing an actor/port mapping.
using unpublish_atom = atom_constant<atom("UNPUBLISH")>;
/// Generic 'CONNECT' atom, e.g., for connecting to remote CAF instances.
using connect_atom = atom_constant<atom("CONNECT")>;
/// Generic 'OPEN' atom, e.g., for opening a port or file.
using open_atom = atom_constant<atom("OPEN")>;
/// Generic 'CLOSE' atom, e.g., for closing a port or file.
using close_atom = atom_constant<atom("CLOSE")>;
} // namespace caf } // namespace caf
namespace std {
template <>
struct hash<caf::atom_value> {
size_t operator()(caf::atom_value x) const {
hash<uint64_t> f;
return f(static_cast<uint64_t>(x));
}
};
} // namespace std
#endif // CAF_ATOM_HPP #endif // CAF_ATOM_HPP
...@@ -20,13 +20,14 @@ ...@@ -20,13 +20,14 @@
#ifndef CAF_DETAIL_ACTOR_REGISTRY_HPP #ifndef CAF_DETAIL_ACTOR_REGISTRY_HPP
#define CAF_DETAIL_ACTOR_REGISTRY_HPP #define CAF_DETAIL_ACTOR_REGISTRY_HPP
#include <map>
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include <atomic> #include <atomic>
#include <cstdint> #include <cstdint>
#include <unordered_map>
#include <condition_variable> #include <condition_variable>
#include "caf/actor.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
...@@ -38,10 +39,8 @@ namespace detail { ...@@ -38,10 +39,8 @@ namespace detail {
class singletons; class singletons;
class actor_registry : public singleton_mixin<actor_registry> { class actor_registry {
public: public:
friend class singleton_mixin<actor_registry>;
~actor_registry(); ~actor_registry();
/// A registry entry consists of a pointer to the actor and an /// A registry entry consists of a pointer to the actor and an
...@@ -82,19 +81,34 @@ public: ...@@ -82,19 +81,34 @@ public:
// blocks the caller until running-actors-count becomes `expected` // blocks the caller until running-actors-count becomes `expected`
void await_running_count_equal(size_t expected); void await_running_count_equal(size_t expected);
actor get_named(atom_value key) const;
std::vector<std::pair<atom_value, actor>> named_actors() const;
static actor_registry* create_singleton();
void dispose();
void stop();
void initialize();
private: private:
using entries = std::map<actor_id, value_type>; using named_entries = std::unordered_map<atom_value, actor>;
using entries = std::unordered_map<actor_id, value_type>;
actor_registry(); actor_registry();
std::atomic<size_t> running_; std::atomic<size_t> running_;
std::atomic<actor_id> ids_;
std::mutex running_mtx_; std::mutex running_mtx_;
std::condition_variable running_cv_; std::condition_variable running_cv_;
mutable detail::shared_spinlock instances_mtx_; mutable detail::shared_spinlock instances_mtx_;
entries entries_; entries entries_;
named_entries named_entries_;
}; };
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_EXPERIMENTAL_ANNOUNCE_ACTOR_HPP
#define CAF_EXPERIMENTAL_ANNOUNCE_ACTOR_HPP
#include <type_traits>
#include "caf/actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
namespace experimental {
enum class spawn_mode {
function,
function_with_selfptr,
clazz
};
template <spawn_mode X>
using spawn_mode_token = std::integral_constant<spawn_mode, X>;
// default: dynamically typed actor without self pointer
template <class Result,
class FirstArg,
bool FirstArgValid =
std::is_base_of<
local_actor,
typename std::remove_pointer<FirstArg>::type
>::value>
struct infer_handle_from_fun_ {
using type = actor;
using impl = event_based_actor;
using behavior_type = behavior;
static constexpr spawn_mode mode = spawn_mode::function;
};
// dynamically typed actor returning a behavior
template <class Impl>
struct infer_handle_from_fun_<void, Impl*, true> {
using type = actor;
using impl = Impl;
using behavior_type = behavior;
static constexpr spawn_mode mode = spawn_mode::function_with_selfptr;
};
// dynamically typed actor with self pointer
template <class Impl>
struct infer_handle_from_fun_<behavior, Impl*, true> {
using type = actor;
using impl = Impl;
using behavior_type = behavior;
static constexpr spawn_mode mode = spawn_mode::function_with_selfptr;
};
// statically typed actor returning a behavior
template <class... Sigs, class FirstArg>
struct infer_handle_from_fun_<typed_behavior<Sigs...>, FirstArg, false> {
using type = typed_actor<Sigs...>;
using impl = typed_event_based_actor<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
static constexpr spawn_mode mode = spawn_mode::function;
};
// statically typed actor with self pointer
template <class Result, class... Sigs>
struct infer_handle_from_fun_<Result, typed_event_based_actor<Sigs...>*, true> {
using type = typed_actor<Sigs...>;
using impl = typed_event_based_actor<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
static constexpr spawn_mode mode = spawn_mode::function_with_selfptr;
};
template <class F, class Trait = typename detail::get_callable_trait<F>::type>
struct infer_handle_from_fun {
using result_type = typename Trait::result_type;
using arg_types = typename Trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using delegate = infer_handle_from_fun_<result_type, first_arg>;
using type = typename delegate::type;
using impl = typename delegate::impl;
using behavior_type = typename delegate::behavior_type;
using fun_type = typename Trait::fun_type;
static constexpr spawn_mode mode = delegate::mode;
};
template <class T>
struct infer_handle_from_behavior {
using type = actor;
};
template <class... Sigs>
struct infer_handle_from_behavior<typed_behavior<Sigs...>> {
using type = typed_actor<Sigs...>;
};
using spawn_result = std::pair<actor_addr, std::set<std::string>>;
using spawn_fun = std::function<spawn_result (message)>;
template <class Trait, class F>
spawn_result dyn_spawn_impl(F ibf) {
using impl = typename Trait::impl;
using behavior_t = typename Trait::behavior_type;
auto ptr = make_counted<impl>();
ptr->initial_behavior_fac([=](local_actor* self) -> behavior {
auto res = ibf(self);
if (res && res->size() > 0 && res->template match_element<behavior_t>(0)) {
return std::move(res->template get_as_mutable<behavior_t>(0).unbox());
}
return {};
});
ptr->launch(nullptr, false, false);
return {ptr->address(), Trait::type::message_types()};
}
template <class Trait, class F>
spawn_result dyn_spawn(F fun, message& msg,
spawn_mode_token<spawn_mode::function>) {
return dyn_spawn_impl<Trait>([=](local_actor*) -> optional<message> {
return const_cast<message&>(msg).apply(fun);
});
}
template <class Trait, class F>
spawn_result dyn_spawn(F fun, message& msg,
spawn_mode_token<spawn_mode::function_with_selfptr>) {
return dyn_spawn_impl<Trait>([=](local_actor* self) -> optional<message> {
// we can't use make_message here because of the implicit conversions
using storage = detail::tuple_vals<typename Trait::impl*>;
auto ptr = make_counted<storage>(static_cast<typename Trait::impl*>(self));
auto m = message{detail::message_data::cow_ptr{std::move(ptr)}} + msg;
return m.apply(fun);
});
}
template <class F>
spawn_fun make_spawn_fun(F fun) {
return [fun](message msg) -> spawn_result {
using trait = infer_handle_from_fun<F>;
spawn_mode_token<trait::mode> tk;
return dyn_spawn<trait>(fun, msg, tk);
};
}
template <class T, class... Ts>
spawn_fun make_spawn_fun() {
static_assert(std::is_same<T*, decltype(new T(std::declval<Ts>()...))>::value,
"no constructor for T(Ts...) exists");
static_assert(detail::conjunction<
std::is_lvalue_reference<Ts>::value...
>::value,
"all Ts must be lvalue references");
static_assert(std::is_base_of<local_actor, T>::value,
"T is not derived from local_actor");
using handle =
typename infer_handle_from_behavior<
typename T::behavior_type
>::type;
using pointer = intrusive_ptr<T>;
auto factory = &make_counted<T, Ts...>;
return [=](message msg) -> spawn_result {
pointer ptr;
auto res = msg.apply(factory);
if (! res || res->empty() || ! res->template match_element<pointer>(0))
return {};
ptr = std::move(res->template get_as_mutable<pointer>(0));
ptr->launch(nullptr, false, false);
return {ptr->address(), handle::message_types()};
};
}
actor spawn_announce_actor_type_server();
void announce_actor_type_impl(std::string&& name, spawn_fun f);
template <class F>
void announce_actor_type(std::string name, F fun) {
announce_actor_type_impl(std::move(name), make_spawn_fun(fun));
}
template <class T, class... Ts>
void announce_actor_type(std::string name) {
announce_actor_type_impl(std::move(name), make_spawn_fun<T, Ts...>());
}
} // namespace experimental
} // namespace caf
#endif // CAF_EXPERIMENTAL_ANNOUNCE_ACTOR_HPP
...@@ -390,6 +390,12 @@ inline message make_message(message other) { ...@@ -390,6 +390,12 @@ inline message make_message(message other) {
return std::move(other); return std::move(other);
} }
/// Returns an empty `message`.
/// @relates message
inline message make_message() {
return message{};
}
/****************************************************************************** /******************************************************************************
* template member function implementations * * template member function implementations *
******************************************************************************/ ******************************************************************************/
......
...@@ -228,6 +228,12 @@ private: ...@@ -228,6 +228,12 @@ private:
behavior bhvr_; behavior bhvr_;
}; };
template <class T>
struct is_typed_behavior : std::false_type { };
template <class... Sigs>
struct is_typed_behavior<typed_behavior<Sigs...>> : std::true_type { };
} // namespace caf } // namespace caf
#endif // CAF_TYPED_BEHAVIOR_HPP #endif // CAF_TYPED_BEHAVIOR_HPP
...@@ -41,7 +41,10 @@ ...@@ -41,7 +41,10 @@
namespace caf { namespace caf {
namespace { namespace {
using guard_type = std::unique_lock<std::mutex>; using guard_type = std::unique_lock<std::mutex>;
std::atomic<actor_id> ids_;
} // namespace <anonymous> } // namespace <anonymous>
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited, // exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
...@@ -59,7 +62,7 @@ abstract_actor::abstract_actor(actor_id aid, node_id nid) ...@@ -59,7 +62,7 @@ abstract_actor::abstract_actor(actor_id aid, node_id nid)
abstract_actor::abstract_actor() abstract_actor::abstract_actor()
: abstract_channel(abstract_channel::is_abstract_actor_flag, : abstract_channel(abstract_channel::is_abstract_actor_flag,
detail::singletons::get_node_id()), detail::singletons::get_node_id()),
id_(detail::singletons::get_actor_registry()->next_id()), id_(++ids_),
exit_reason_(exit_reason::not_exited), exit_reason_(exit_reason::not_exited),
host_(nullptr) { host_(nullptr) {
// nop // nop
......
...@@ -23,10 +23,14 @@ ...@@ -23,10 +23,14 @@
#include <limits> #include <limits>
#include <stdexcept> #include <stdexcept>
#include "caf/spawn.hpp"
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#include "caf/scheduler/detached_threads.hpp" #include "caf/scheduler/detached_threads.hpp"
...@@ -47,7 +51,7 @@ actor_registry::~actor_registry() { ...@@ -47,7 +51,7 @@ actor_registry::~actor_registry() {
// nop // nop
} }
actor_registry::actor_registry() : running_(0), ids_(1) { actor_registry::actor_registry() : running_(0) {
// nop // nop
} }
...@@ -95,10 +99,6 @@ void actor_registry::erase(actor_id key, uint32_t reason) { ...@@ -95,10 +99,6 @@ void actor_registry::erase(actor_id key, uint32_t reason) {
} }
} }
uint32_t actor_registry::next_id() {
return ++ids_;
}
void actor_registry::inc_running() { void actor_registry::inc_running() {
# if defined(CAF_LOG_LEVEL) && CAF_LOG_LEVEL >= CAF_DEBUG # if defined(CAF_LOG_LEVEL) && CAF_LOG_LEVEL >= CAF_DEBUG
CAF_LOG_DEBUG("new value = " << ++running_); CAF_LOG_DEBUG("new value = " << ++running_);
...@@ -130,5 +130,46 @@ void actor_registry::await_running_count_equal(size_t expected) { ...@@ -130,5 +130,46 @@ void actor_registry::await_running_count_equal(size_t expected) {
} }
} }
actor actor_registry::get_named(atom_value key) const {
auto i = named_entries_.find(key);
if (i == named_entries_.end())
return invalid_actor;
return i->second;
}
std::vector<std::pair<atom_value, actor>> actor_registry::named_actors() const {
std::vector<std::pair<atom_value, actor>> result;
for (auto& kvp : named_entries_)
result.emplace_back(kvp);
return result;
}
actor_registry* actor_registry::create_singleton() {
return new actor_registry;
}
void actor_registry::dispose() {
delete this;
}
void actor_registry::stop() {
scoped_actor self{true};
for (auto& kvp : named_entries_) {
self->monitor(kvp.second);
self->send_exit(kvp.second, exit_reason::kill);
self->receive(
[](const down_msg&) {
// nop
}
);
}
named_entries_.clear();
}
void actor_registry::initialize() {
named_entries_.emplace(atom("spawner"),
experimental::spawn_announce_actor_type_server());
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/experimental/announce_actor_type.hpp"
#include "caf/atom.hpp"
#include "caf/send.hpp"
#include "caf/spawn.hpp"
#include "caf/to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/experimental/stateful_actor.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
namespace experimental {
namespace {
struct spawner_state {
std::unordered_map<std::string, spawn_fun> funs_;
};
behavior announce_actor_type_server(stateful_actor<spawner_state>* self) {
return {
[=](add_atom, std::string& name, spawn_fun& f) {
self->state.funs_.emplace(std::move(name), std::move(f));
},
[=](get_atom, const std::string& name, message& args) -> spawn_result {
auto i = self->state.funs_.find(name);
if (i == self->state.funs_.end())
return std::make_pair(invalid_actor_addr, std::set<std::string>{});
auto f = i->second;
return f(args);
},
others >> [=] {
CAF_LOGF_WARNING("Unexpected message: "
<< to_string(self->current_message()));
}
};
}
} // namespace <anonymous>
actor spawn_announce_actor_type_server() {
return spawn<hidden + lazy_init>(announce_actor_type_server);
}
void announce_actor_type_impl(std::string&& name, spawn_fun f) {
auto registry = detail::singletons::get_actor_registry();
auto server = registry->get_named(atom("spawner"));
anon_send(server, add_atom::value, std::move(name), std::move(f));
}
} // namespace experimental
} // namespace caf
...@@ -76,12 +76,12 @@ void singletons::stop_singletons() { ...@@ -76,12 +76,12 @@ void singletons::stop_singletons() {
} }
CAF_LOGF_DEBUG("stop group manager"); CAF_LOGF_DEBUG("stop group manager");
stop(s_group_manager); stop(s_group_manager);
CAF_LOGF_DEBUG("stop actor registry");
stop(s_actor_registry);
CAF_LOGF_DEBUG("stop scheduler"); CAF_LOGF_DEBUG("stop scheduler");
stop(s_scheduling_coordinator); stop(s_scheduling_coordinator);
CAF_LOGF_DEBUG("wait for all detached threads"); CAF_LOGF_DEBUG("wait for all detached threads");
scheduler::await_detached_threads(); scheduler::await_detached_threads();
CAF_LOGF_DEBUG("stop actor registry");
stop(s_actor_registry);
// dispose singletons, i.e., release memory // dispose singletons, i.e., release memory
CAF_LOGF_DEBUG("dispose plugins"); CAF_LOGF_DEBUG("dispose plugins");
for (auto& plugin : s_plugins) { for (auto& plugin : s_plugins) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](spawn_result& res) {
if (expect_fail) {
CAF_REQUIRE(res.first == invalid_actor_addr);
return;
}
CAF_REQUIRE(res.first != invalid_actor_addr);
CAF_CHECK(res.second.empty());
aut = actor_cast<actor>(res.first);
},
others >> [&] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
}
);
}
~fixture() {
{ // lifetime scope of scoped_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();
}
};
} // namespace <anonymous>
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()
...@@ -30,6 +30,67 @@ using std::endl; ...@@ -30,6 +30,67 @@ using std::endl;
namespace { namespace {
struct meta_info {
constexpr meta_info(const char* x, size_t y, uint32_t z)
: name(x), hash(y), version(z) {
// nop
}
constexpr meta_info(const meta_info& other)
: name(other.name),
hash(other.hash),
version(other.version) {
// nop
}
const char* name;
size_t hash;
uint32_t version;
};
constexpr size_t str_hash(const char* cstr, size_t interim = 0) {
return (*cstr == '\0') ? interim : str_hash(cstr + 1, interim * 101 + *cstr);
}
constexpr meta_info make_meta_info(const char* name, uint32_t version) {
return meta_info{name, str_hash(name), version};
}
template <class T>
struct meta_information;
#define CAF_META_INFORMATION(class_name, version) \
template <> \
struct meta_information<class_name> { \
static constexpr meta_info value = make_meta_info(#class_name, version); \
}; \
constexpr meta_info meta_information<class_name>::value;
#define DUMMY(name) \
class name {}; \
CAF_META_INFORMATION(name, 0)
DUMMY(foo1)
DUMMY(foo2)
DUMMY(foo3)
DUMMY(foo4)
DUMMY(foo5)
DUMMY(foo6)
DUMMY(foo7)
DUMMY(foo8)
DUMMY(foo9)
DUMMY(foo10)
DUMMY(foo11)
DUMMY(foo12)
DUMMY(foo13)
DUMMY(foo14)
DUMMY(foo15)
DUMMY(foo16)
DUMMY(foo17)
DUMMY(foo18)
DUMMY(foo19)
DUMMY(foo20)
struct fixture { struct fixture {
~fixture() { ~fixture() {
await_all_actors_done(); await_all_actors_done();
...@@ -37,6 +98,7 @@ struct fixture { ...@@ -37,6 +98,7 @@ struct fixture {
} }
}; };
constexpr const char* global_redirect = ":test"; constexpr const char* global_redirect = ":test";
constexpr const char* local_redirect = ":test2"; constexpr const char* local_redirect = ":test2";
...@@ -57,6 +119,133 @@ void chattier_actor(event_based_actor* self, const std::string& fn) { ...@@ -57,6 +119,133 @@ void chattier_actor(event_based_actor* self, const std::string& fn) {
CAF_TEST_FIXTURE_SCOPE(aout_tests, fixture) CAF_TEST_FIXTURE_SCOPE(aout_tests, fixture)
const meta_info* lookup(const std::vector<std::pair<size_t, const meta_info*>>& haystack,
const meta_info* needle) {
/*
auto h = needle->hash;
auto e = haystack.end();
for (auto i = haystack.begin(); i != e; ++i) {
if (i->first == h) {
auto n = i + 1;
if (n == e || n->second->hash != h)
return i->second;
for (auto j = i; j != e; ++j)
if (strcmp(j->second->name, needle->name) == 0)
return j->second;
}
}
*/
auto h = needle->hash;
auto e = haystack.end();
//auto i = std::find_if(haystack.begin(), e,
// [h](const std::pair<size_t, const meta_info*>& x ) { return x.first == h; });
auto i = std::lower_bound(haystack.begin(), e, h,
[](const std::pair<size_t, const meta_info*>& x, size_t y) { return x.first < y; });
if (i == e || i->first != h)
return nullptr;
// check for collision
auto n = i + 1;
if (n == e || n->first != h)
return i->second;
i = std::find_if(i, e, [needle](const std::pair<size_t, const meta_info*>& x ) { return strcmp(x.second->name, needle->name) == 0; });
if (i != e)
return i->second;
return nullptr;
}
const meta_info* lookup(std::unordered_multimap<size_t, const meta_info*>& haystack,
const meta_info* needle) {
auto i = haystack.find(needle->hash);
auto e = haystack.end();
if (i == e)
return nullptr;
auto n = std::next(i);
if (n == e || n->second->hash != i->second->hash)
return i->second;
for (; i != e; ++i)
if (strcmp(i->second->name, needle->name) == 0)
return i->second;
return nullptr;
}
const meta_info* lookup(std::multimap<size_t, const meta_info*>& haystack,
const meta_info* needle) {
auto i = haystack.find(needle->hash);
auto e = haystack.end();
if (i == e)
return nullptr;
auto n = std::next(i);
if (n == e || n->second->hash != i->second->hash)
return i->second;
for (; i != e; ++i)
if (strcmp(i->second->name, needle->name) == 0)
return i->second;
return nullptr;
}
CAF_TEST(foobar) {
using std::make_pair;
std::vector<std::pair<size_t, const meta_info*>> map1;
std::unordered_multimap<size_t, const meta_info*> map2;
std::multimap<size_t, const meta_info*> map3;
const meta_info* arr[] = {
&meta_information<foo1>::value,
&meta_information<foo2>::value,
&meta_information<foo3>::value,
&meta_information<foo4>::value,
&meta_information<foo5>::value,
&meta_information<foo6>::value,
&meta_information<foo7>::value,
&meta_information<foo8>::value,
&meta_information<foo9>::value,
&meta_information<foo10>::value,
&meta_information<foo11>::value,
&meta_information<foo12>::value,
&meta_information<foo13>::value,
&meta_information<foo14>::value,
&meta_information<foo15>::value,
&meta_information<foo16>::value,
&meta_information<foo17>::value,
&meta_information<foo18>::value,
&meta_information<foo19>::value,
&meta_information<foo20>::value
};
for (auto i = std::begin(arr); i != std::end(arr); ++i) {
map1.emplace_back((*i)->hash, *i);
map2.emplace((*i)->hash, *i);
map3.emplace((*i)->hash, *i);
}
std::sort(map1.begin(), map1.end());
std::array<const meta_info*, 20> dummy;
{
auto t0 = std::chrono::high_resolution_clock::now();
for (auto i = 0; i < 10000; ++i)
dummy[i % 20] = lookup(map1, arr[i % 20]);
auto t1 = std::chrono::high_resolution_clock::now();
std::cout << "vector: " << std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count() << std::endl;
std::cout << "check: " << std::equal(dummy.begin(), dummy.end(), std::begin(arr)) << std::endl;
}
{
auto t0 = std::chrono::high_resolution_clock::now();
for (auto i = 0; i < 10000; ++i)
dummy[i % 20] = lookup(map2, arr[i % 20]);
auto t1 = std::chrono::high_resolution_clock::now();
std::cout << "hash map: " << std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count() << std::endl;
std::cout << "check: " << std::equal(dummy.begin(), dummy.end(), std::begin(arr)) << std::endl;
}
{
auto t0 = std::chrono::high_resolution_clock::now();
for (auto i = 0; i < 10000; ++i)
dummy[i % 20] = lookup(map3, arr[i % 20]);
auto t1 = std::chrono::high_resolution_clock::now();
std::cout << "map: " << std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count() << std::endl;
std::cout << "check: " << std::equal(dummy.begin(), dummy.end(), std::begin(arr)) << std::endl;
}
CAF_CHECK(meta_information<foo1>::value.name == "foo1");
CAF_CHECK(meta_information<foo1>::value.hash == str_hash("foo1"));
CAF_CHECK(meta_information<foo1>::value.version == 0);
}
CAF_TEST(global_redirect) { CAF_TEST(global_redirect) {
scoped_actor self; scoped_actor self;
self->join(group::get("local", global_redirect)); self->join(group::get("local", global_redirect));
......
...@@ -27,6 +27,18 @@ ...@@ -27,6 +27,18 @@
using namespace caf; using namespace caf;
CAF_TEST(apply) {
auto f1 = [] {
CAF_TEST_ERROR("f1 invoked!");
};
auto f2 = [](int i) {
CAF_CHECK_EQUAL(i, 42);
};
auto m = make_message(42);
m.apply(f1);
m.apply(f2);
}
CAF_TEST(drop) { CAF_TEST(drop) {
auto m1 = make_message(1, 2, 3, 4, 5); auto m1 = make_message(1, 2, 3, 4, 5);
std::vector<message> messages{ std::vector<message> messages{
......
...@@ -369,7 +369,7 @@ public: ...@@ -369,7 +369,7 @@ public:
/// Called if a server handshake was received and /// Called if a server handshake was received and
/// the connection to `nid` is established. /// the connection to `nid` is established.
virtual void finalize_handshake(const node_id& nid, actor_id aid, virtual void finalize_handshake(const node_id& nid, actor_id aid,
const std::set<std::string>& sigs) = 0; std::set<std::string>& sigs) = 0;
/// Called whenever a direct connection was closed or a /// Called whenever a direct connection was closed or a
/// node became unrechable for other reasons *before* /// node became unrechable for other reasons *before*
......
...@@ -49,7 +49,7 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee { ...@@ -49,7 +49,7 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// inherited from basp::instance::listener // inherited from basp::instance::listener
void finalize_handshake(const node_id& nid, actor_id aid, void finalize_handshake(const node_id& nid, actor_id aid,
const std::set<std::string>& sigs) override; std::set<std::string>& sigs) override;
// inherited from basp::instance::listener // inherited from basp::instance::listener
void purge_state(const node_id& id) override; void purge_state(const node_id& id) override;
...@@ -71,7 +71,6 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee { ...@@ -71,7 +71,6 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
node_id id; node_id id;
uint16_t remote_port; uint16_t remote_port;
optional<response_promise> callback; optional<response_promise> callback;
std::set<std::string> expected_sigs;
}; };
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
......
...@@ -30,99 +30,78 @@ namespace io { ...@@ -30,99 +30,78 @@ namespace io {
/// ///
/// The interface implements the following pseudo code. /// The interface implements the following pseudo code.
/// ~~~ /// ~~~
/// using put_result =
/// either (ok_atom, uint16_t port)
/// or (error_atom, string error_string);
///
/// using get_result =
/// either (ok_atom, actor_addr remote_address)
/// or (error_atom, string error_string);
///
/// using delete_result =
/// either (ok_atom)
/// or (error_atom, string error_string);
///
/// interface middleman_actor { /// interface middleman_actor {
/// (put_atom, actor_addr whom, uint16_t port, string addr, bool reuse_addr)
/// -> put_result;
/// ///
/// (put_atom, actor_addr whom, uint16_t port, string addr) /// // Establishes a new `port <-> actor` mapping and returns the actual
/// -> put_result; /// // port in use on success. Passing 0 as port instructs the OS to choose
/// // the next high-level port available for binding.
/// // @param port: Unused TCP port or 0 for any.
/// // @param whom: Actor that should be published at given port.
/// // @param ifs: Interface of given actor.
/// // @param addr: IP address to listen to or empty for any.
/// // @param reuse_addr: Enables or disables SO_REUSEPORT option.
/// (publish_atom, uint16_t port, actor_addr whom,
/// set<string> ifs, string addr, bool reuse_addr)
/// -> either (ok_atom, uint16_t port)
/// or (error_atom, string error_string)
/// ///
/// (put_atom, actor_addr whom, uint16_t port, bool reuse_addr) /// // Opens a new port other CAF instances can connect to. The
/// -> put_result; /// // difference between `PUBLISH` and `OPEN` is that no actor is mapped to
/// // this port, meaning that connecting nodes only get a valid `node_id`
/// // handle when connecting.
/// // @param port: Unused TCP port or 0 for any.
/// // @param addr: IP address to listen to or empty for any.
/// // @param reuse_addr: Enables or disables SO_REUSEPORT option.
/// (open_atom, uint16_t port, string addr, bool reuse_addr)
/// -> either (ok_atom, uint16_t port)
/// or (error_atom, string error_string)
/// ///
/// (put_atom, actor_addr whom, uint16_t port) /// // Queries a remote node and returns an ID to this node as well as
/// -> put_result; /// // an `actor_addr` to a remote actor if an actor was published at this
/// // port. The actor address must be cast to either `actor` or
/// // `typed_actor` using `actor_cast` after validating `ifs`.
/// // @param hostname IP address or DNS hostname.
/// // @param port TCP port.
/// (connect_atom, string hostname, uint16_t port)
/// -> either (ok_atom, node_id nid, actor_addr remote_actor, set<string> ifs)
/// or (error_atom, string error_string)
/// ///
/// (get_atom, string hostname, uint16_t port) /// // Closes `port` if it is mapped to `whom`.
/// -> get_result; /// // @param whom A published actor.
/// // @param port Used TCP port.
/// (unpublish_atom, actor_addr whom, uint16_t port)
/// -> either (ok_atom)
/// or (error_atom, string error_string)
/// ///
/// (get_atom, string hostname, uint16_t port, set<string> expected_ifs) /// // Unconditionally closes `port`, removing any actor
/// -> get_result; /// // published at this port.
/// // @param port Used TCP port.
/// (close_atom, uint16_t port)
/// -> either (ok_atom)
/// or (error_atom, string error_string)
/// ///
/// (delete_atom, actor_addr whom)
/// -> delete_result;
///
/// (delete_atom, actor_addr whom, uint16_t port)
/// -> delete_result;
/// } /// }
/// ~~~ /// ~~~
///
/// The `middleman_actor` actor offers the following operations:
/// - `PUT` establishes a new `port <-> actor`
/// mapping and returns the actual port in use on success.
/// Passing 0 as port instructs the OS to choose the next high-level port
/// available for binding.
/// Type | Name | Parameter Description
/// -----------|------------|--------------------------------------------------
/// put_atom | | Identifies `PUT` operations.
/// actor_addr | whom | Actor that should be published at given port.
/// uint16_t | port | Unused TCP port or 0 for any.
/// string | addr | Optional; IP address to listen to or `INADDR_ANY`
/// bool | reuse_addr | Optional; enable SO_REUSEPORT option
///
/// - `GET` queries a remote node and returns an `actor_addr` to the remote actor
/// on success. This handle must be cast to either `actor` or `typed_actor`
/// using `actor_cast`.
/// Type | Name | Parameter Description
/// ------------|-------------|------------------------------------------------
/// get_atom | | Identifies `GET` operations.
/// string | hostname | Valid hostname or IP address.
/// uint16_t | port | TCP port.
/// set<string> | expected_ifs | Optional; Interface of typed remote actor.
///
/// - `DELETE` removes either all `port <-> actor` mappings for an actor or only
/// a single one if the optional `port` parameter is set.
/// Type | Name | Parameter Description
/// ------------|-------------|------------------------------------------------
/// delete_atom | | Identifies `DELETE` operations.
/// actor_addr | whom | Published actor.
/// uint16_t | port | Optional; remove only a single mapping.
using middleman_actor = using middleman_actor =
typed_actor< typed_actor<
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>, std::string, bool> replies_to<publish_atom, uint16_t, actor_addr,
::with_either<ok_atom, uint16_t> std::set<std::string>, std::string, bool>
::or_else<error_atom, std::string>,
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>, std::string>
::with_either<ok_atom, uint16_t> ::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>, ::or_else<error_atom, std::string>,
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>, bool>
::with_either<ok_atom, uint16_t> replies_to<open_atom, uint16_t, std::string, bool>
::or_else<error_atom, std::string>,
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>>
::with_either<ok_atom, uint16_t> ::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>, ::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t>
::with_either<ok_atom, actor_addr> replies_to<connect_atom, std::string, uint16_t>
::or_else<error_atom, std::string>, ::with_either<ok_atom, node_id, actor_addr, std::set<std::string>>
replies_to<get_atom, std::string, uint16_t, std::set<std::string>>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>, ::or_else<error_atom, std::string>,
replies_to<delete_atom, actor_addr>
replies_to<unpublish_atom, actor_addr, uint16_t>
::with_either<ok_atom> ::with_either<ok_atom>
::or_else<error_atom, std::string>, ::or_else<error_atom, std::string>,
replies_to<delete_atom, actor_addr, uint16_t>
replies_to<close_atom, uint16_t>
::with_either<ok_atom> ::with_either<ok_atom>
::or_else<error_atom, std::string>>; ::or_else<error_atom, std::string>>;
...@@ -133,4 +112,3 @@ middleman_actor get_middleman_actor(); // implemented in middleman.cpp ...@@ -133,4 +112,3 @@ middleman_actor get_middleman_actor(); // implemented in middleman.cpp
} // namespace caf } // namespace caf
#endif // CAF_IO_MIDDLEMAN_ACTOR_HPP #endif // CAF_IO_MIDDLEMAN_ACTOR_HPP
...@@ -31,8 +31,8 @@ ...@@ -31,8 +31,8 @@
namespace caf { namespace caf {
namespace io { namespace io {
abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs, actor_addr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port); std::string host, uint16_t port);
/// Establish a new connection to the actor at `host` on given `port`. /// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address. /// @param host Valid hostname or IP address.
...@@ -41,8 +41,8 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs, ...@@ -41,8 +41,8 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
/// representing a remote actor. /// representing a remote actor.
/// @throws network_error Thrown on connection error or /// @throws network_error Thrown on connection error or
/// when connecting to a typed actor. /// when connecting to a typed actor.
inline actor remote_actor(const std::string& host, uint16_t port) { inline actor remote_actor(std::string host, uint16_t port) {
auto res = remote_actor_impl(std::set<std::string>{}, host, port); auto res = remote_actor_impl(std::set<std::string>{}, std::move(host), port);
return actor_cast<actor>(res); return actor_cast<actor>(res);
} }
...@@ -54,10 +54,10 @@ inline actor remote_actor(const std::string& host, uint16_t port) { ...@@ -54,10 +54,10 @@ inline actor remote_actor(const std::string& host, uint16_t port) {
/// @throws network_error Thrown on connection error or when connecting /// @throws network_error Thrown on connection error or when connecting
/// to an untyped otherwise unexpected actor. /// to an untyped otherwise unexpected actor.
template <class ActorHandle> template <class ActorHandle>
ActorHandle typed_remote_actor(const std::string& host, uint16_t port) { ActorHandle typed_remote_actor(std::string host, uint16_t port) {
auto iface = ActorHandle::message_types(); auto res = remote_actor_impl(ActorHandle::message_types(),
return actor_cast<ActorHandle>(remote_actor_impl(std::move(iface), std::move(host), port);
host, port)); return actor_cast<ActorHandle>(res);
} }
} // namespace io } // namespace io
......
...@@ -93,7 +93,7 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid, ...@@ -93,7 +93,7 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
} }
void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
const std::set<std::string>& sigs) { std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_TSARG(nid) CAF_LOG_TRACE(CAF_TSARG(nid)
<< ", " << CAF_ARG(aid) << ", " << CAF_ARG(aid)
<< ", " << CAF_TSARG(make_message(sigs))); << ", " << CAF_TSARG(make_message(sigs)));
...@@ -102,20 +102,15 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, ...@@ -102,20 +102,15 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
auto& cb = this_context->callback; auto& cb = this_context->callback;
if (! cb) if (! cb)
return; return;
auto& exp = this_context->expected_sigs;
auto cleanup = detail::make_scope_guard([&] { auto cleanup = detail::make_scope_guard([&] {
cb = none; cb = none;
exp.clear();
}); });
if (! std::includes(sigs.begin(), sigs.end(), exp.begin(), exp.end())) {
cb->deliver(make_message(error_atom::value,
"expected signature does not "
"comply to found signature"));
return;
}
if (aid == invalid_actor_id) { if (aid == invalid_actor_id) {
// can occur when connecting to the default port of a node // can occur when connecting to the default port of a node
cb->deliver(make_message(ok_atom::value, actor_addr{invalid_actor_addr})); cb->deliver(make_message(ok_atom::value,
nid,
actor_addr{invalid_actor_addr},
std::move(sigs)));
return; return;
} }
abstract_actor_ptr ptr; abstract_actor_ptr ptr;
...@@ -132,9 +127,8 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, ...@@ -132,9 +127,8 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
actor_addr addr = ptr ? ptr->address() : invalid_actor_addr; actor_addr addr = ptr ? ptr->address() : invalid_actor_addr;
if (addr.is_remote()) if (addr.is_remote())
known_remotes.emplace(nid, std::make_pair(this_context->remote_port, addr)); known_remotes.emplace(nid, std::make_pair(this_context->remote_port, addr));
cb->deliver(make_message(ok_atom::value, addr)); cb->deliver(make_message(ok_atom::value, nid, addr, std::move(sigs)));
this_context->callback = none; this_context->callback = none;
this_context->expected_sigs.clear();
} }
void basp_broker_state::purge_state(const node_id& nid) { void basp_broker_state::purge_state(const node_id& nid) {
...@@ -236,8 +230,7 @@ void basp_broker_state::set_context(connection_handle hdl) { ...@@ -236,8 +230,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
hdl, hdl,
invalid_node_id, invalid_node_id,
0, 0,
none, none}).first;
std::set<std::string>{}}).first;
} }
this_context = &i->second; this_context = &i->second;
} }
...@@ -339,7 +332,7 @@ behavior basp_broker::make_behavior() { ...@@ -339,7 +332,7 @@ behavior basp_broker::make_behavior() {
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
}, },
// received from middleman actor // received from middleman actor
[=](put_atom, accept_handle hdl, uint16_t port, const actor_addr& whom, [=](publish_atom, accept_handle hdl, uint16_t port, const actor_addr& whom,
std::set<std::string>& sigs) { std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom) CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom)
<< ", " << CAF_ARG(port)); << ", " << CAF_ARG(port));
...@@ -357,8 +350,7 @@ behavior basp_broker::make_behavior() { ...@@ -357,8 +350,7 @@ behavior basp_broker::make_behavior() {
parent().notify<hook::actor_published>(whom, port); parent().notify<hook::actor_published>(whom, port);
}, },
// received from middleman actor (delegated) // received from middleman actor (delegated)
[=](get_atom, connection_handle hdl, uint16_t port, [=](connect_atom, connection_handle hdl, uint16_t port) {
std::set<std::string>& expected_ifs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
auto rp = make_response_promise(); auto rp = make_response_promise();
try { try {
...@@ -376,7 +368,6 @@ behavior basp_broker::make_behavior() { ...@@ -376,7 +368,6 @@ behavior basp_broker::make_behavior() {
ctx.remote_port = port; ctx.remote_port = port;
ctx.cstate = basp::await_header; ctx.cstate = basp::await_header;
ctx.callback = rp; ctx.callback = rp;
ctx.expected_sigs = std::move(expected_ifs);
// await server handshake // await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size)); configure_read(hdl, receive_policy::exactly(basp::header_size));
}, },
...@@ -384,7 +375,7 @@ behavior basp_broker::make_behavior() { ...@@ -384,7 +375,7 @@ behavior basp_broker::make_behavior() {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid));
state.get_namespace().erase(nid, aid); state.get_namespace().erase(nid, aid);
}, },
[=](delete_atom, const actor_addr& whom, uint16_t port) -> message { [=](unpublish_atom, const actor_addr& whom, uint16_t port) -> message {
CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port)); CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port));
if (whom == invalid_actor_addr) if (whom == invalid_actor_addr)
return make_message(error_atom::value, "whom == invalid_actor_addr"); return make_message(error_atom::value, "whom == invalid_actor_addr");
...@@ -397,6 +388,19 @@ behavior basp_broker::make_behavior() { ...@@ -397,6 +388,19 @@ behavior basp_broker::make_behavior() {
? make_message(error_atom::value, "no mapping found") ? make_message(error_atom::value, "no mapping found")
: make_message(ok_atom::value); : make_message(ok_atom::value);
}, },
[=](close_atom, uint16_t port) -> message {
if (port == 0)
return make_message(error_atom::value, "port == 0");
// it is well-defined behavior to not have an actor published here,
// hence the result can be ignored safely
state.instance.remove_published_actor(port, nullptr);
auto acceptor = hdl_by_port(port);
if (acceptor) {
close(*acceptor);
return make_message(ok_atom::value);
}
return make_message(error_atom::value, "no doorman for given port found");
},
// catch-all error handler // catch-all error handler
others >> others >>
[=] { [=] {
......
...@@ -139,8 +139,6 @@ void do_announce(const char* tname) { ...@@ -139,8 +139,6 @@ void do_announce(const char* tname) {
announce(typeid(T), uniform_type_info_ptr{new uti_impl<T>(tname)}); announce(typeid(T), uniform_type_info_ptr{new uti_impl<T>(tname)});
} }
} // namespace <anonymous>
class middleman_actor_impl : public middleman_actor::base { class middleman_actor_impl : public middleman_actor::base {
public: public:
middleman_actor_impl(middleman& mref, actor default_broker) middleman_actor_impl(middleman& mref, actor default_broker)
...@@ -149,65 +147,62 @@ public: ...@@ -149,65 +147,62 @@ public:
// nop // nop
} }
~middleman_actor_impl();
void on_exit() { void on_exit() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
broker_ = invalid_actor; broker_ = invalid_actor;
} }
using get_op_result = either<ok_atom, actor_addr> using put_res = either<ok_atom, uint16_t>::or_else<error_atom, std::string>;
::or_else<error_atom, std::string>;
using get_op_promise = typed_response_promise<get_op_result>;
using del_op_result = either<ok_atom>::or_else<error_atom, std::string>; using get_res = delegated<either<ok_atom, node_id, actor_addr,
std::set<std::string>>
::or_else<error_atom, std::string>>;
using del_op_promise = delegated<del_op_result>; using del_res = delegated<either<ok_atom>::or_else<error_atom, std::string>>;
using map_type = std::map<int64_t, response_promise>;
behavior_type make_behavior() override { behavior_type make_behavior() override {
return { return {
[=](put_atom, uint16_t port, const actor_addr& whom, [=](publish_atom, uint16_t port, actor_addr& whom,
std::set<std::string>& sigs, const std::string& addr, std::set<std::string>& sigs, std::string& addr, bool reuse) {
bool reuse_addr) { return put(port, whom, sigs, addr.c_str(), reuse);
return put(port, whom, sigs, addr.c_str(), reuse_addr);
},
[=](put_atom, uint16_t port, const actor_addr& whom,
std::set<std::string>& sigs, const std::string& addr) {
return put(port, whom, sigs, addr.c_str());
},
[=](put_atom, uint16_t port, const actor_addr& whom,
std::set<std::string>& sigs, bool reuse_addr) {
return put(port, whom, sigs, nullptr, reuse_addr);
}, },
[=](put_atom, uint16_t port, const actor_addr& whom, [=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
std::set<std::string>& sigs) { actor_addr whom = invalid_actor_addr;
return put(port, whom, sigs); std::set<std::string> sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
}, },
[=](get_atom, const std::string& hostname, uint16_t port, [=](connect_atom, const std::string& hostname, uint16_t port) -> get_res {
std::set<std::string>& expected_ifs) { CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
return get(hostname, port, std::move(expected_ifs)); try {
auto hdl = parent_.backend().new_tcp_scribe(hostname, port);
delegate(broker_, connect_atom::value, hdl, port);
}
catch (network_error& err) {
// fullfil promise immediately
std::string msg = "network_error: ";
msg += err.what();
auto rp = make_response_promise();
rp.deliver(make_message(error_atom::value, std::move(msg)));
}
return {};
}, },
[=](get_atom, const std::string& hostname, uint16_t port) { [=](unpublish_atom, actor_addr& whom, uint16_t port) -> del_res {
return get(hostname, port, std::set<std::string>()); delegate(broker_, unpublish_atom::value, std::move(whom), port);
return {};
}, },
[=](delete_atom, const actor_addr& whom) { [=](close_atom, uint16_t port) -> del_res {
return del(whom); delegate(broker_, close_atom::value, port);
}, return {};
[=](delete_atom, const actor_addr& whom, uint16_t port) {
return del(whom, port);
} }
}; };
} }
private: private:
either<ok_atom, uint16_t>::or_else<error_atom, std::string> put_res put(uint16_t port, actor_addr& whom,
put(uint16_t port, const actor_addr& whom, std::set<std::string>& sigs, std::set<std::string>& sigs, const char* in = nullptr,
const char* in = nullptr, bool reuse_addr = false) { bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port) CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port) << ", "
<< ", " << CAF_ARG(reuse_addr)); << CAF_ARG(reuse_addr));
accept_handle hdl; accept_handle hdl;
uint16_t actual_port; uint16_t actual_port;
try { try {
...@@ -225,66 +220,16 @@ private: ...@@ -225,66 +220,16 @@ private:
catch (network_error& err) { catch (network_error& err) {
return {error_atom::value, std::string("network_error: ") + err.what()}; return {error_atom::value, std::string("network_error: ") + err.what()};
} }
send(broker_, put_atom::value, hdl, actual_port, whom, std::move(sigs)); send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
return {ok_atom::value, actual_port}; return {ok_atom::value, actual_port};
} }
get_op_promise get(const std::string& hostname, uint16_t port,
std::set<std::string> expected_ifs) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
get_op_promise result;
try {
auto hdl = parent_.backend().new_tcp_scribe(hostname, port);
delegate(broker_, get_atom::value, hdl, port, std::move(expected_ifs));
}
catch (network_error& err) {
// fullfil promise immediately
std::string msg = "network_error: ";
msg += err.what();
result = make_response_promise();
result.deliver(get_op_result{error_atom::value, std::move(msg)});
}
return result;
}
del_op_promise del(const actor_addr& whom, uint16_t port = 0) {
CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port));
delegate(broker_, delete_atom::value, whom, port);
return {};
}
template <class T, class... Ts>
void handle_ok(map_type& storage, int64_t request_id, Ts&&... xs) {
CAF_LOG_TRACE(CAF_ARG(request_id));
auto i = storage.find(request_id);
if (i == storage.end()) {
CAF_LOG_ERROR("request id not found: " << request_id);
return;
}
i->second.deliver(T{ok_atom::value, std::forward<Ts>(xs)...}.value);
storage.erase(i);
}
template <class F>
bool finalize_request(map_type& storage, int64_t req_id, F fun) {
CAF_LOG_TRACE(CAF_ARG(req_id));
auto i = storage.find(req_id);
if (i == storage.end()) {
CAF_LOG_INFO("request ID not found in storage");
return false;
}
fun(i->second);
storage.erase(i);
return true;
}
actor broker_; actor broker_;
middleman& parent_; middleman& parent_;
}; };
middleman_actor_impl::~middleman_actor_impl() { } // namespace <anonymous>
CAF_LOG_TRACE("");
}
middleman* middleman::instance() { middleman* middleman::instance() {
CAF_LOGF_TRACE(""); CAF_LOGF_TRACE("");
......
...@@ -53,7 +53,7 @@ uint16_t publish_impl(uint16_t port, actor_addr whom, ...@@ -53,7 +53,7 @@ uint16_t publish_impl(uint16_t port, actor_addr whom,
uint16_t result; uint16_t result;
std::string error_msg; std::string error_msg;
try { try {
self->sync_send(mm, put_atom::value, port, self->sync_send(mm, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).await( std::move(whom), std::move(sigs), str, ru).await(
[&](ok_atom, uint16_t res) { [&](ok_atom, uint16_t res) {
result = res; result = res;
......
...@@ -42,14 +42,20 @@ ...@@ -42,14 +42,20 @@
namespace caf { namespace caf {
namespace io { namespace io {
abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs, actor_addr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port) { std::string host, uint16_t port) {
auto mm = get_middleman_actor(); auto mm = get_middleman_actor();
actor_addr result;
scoped_actor self; scoped_actor self;
abstract_actor_ptr result; self->sync_send(mm, connect_atom::value, std::move(host), port).await(
self->sync_send(mm, get_atom{}, std::move(host), port, std::move(ifs)).await( [&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) {
[&](ok_atom, actor_addr res) { if (!res)
result = actor_cast<abstract_actor_ptr>(res); throw network_error("no actor published at given port");
if (! (xs.empty() && ifs.empty())
&& ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end()))
throw network_error("expected signature does not "
"comply to found signature");
result = std::move(res);
}, },
[&](error_atom, std::string& msg) { [&](error_atom, std::string& msg) {
throw network_error(std::move(msg)); throw network_error(std::move(msg));
...@@ -60,4 +66,3 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs, ...@@ -60,4 +66,3 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -34,7 +34,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) { ...@@ -34,7 +34,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) {
auto mm = get_middleman_actor(); auto mm = get_middleman_actor();
if (blocking) { if (blocking) {
scoped_actor self; scoped_actor self;
self->sync_send(mm, delete_atom::value, whom, port).await( self->sync_send(mm, unpublish_atom::value, whom, port).await(
[](ok_atom) { [](ok_atom) {
// ok, basp_broker is done // ok, basp_broker is done
}, },
...@@ -43,7 +43,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) { ...@@ -43,7 +43,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) {
} }
); );
} else { } else {
anon_send(mm, delete_atom::value, whom, port); anon_send(mm, unpublish_atom::value, whom, port);
} }
} }
......
...@@ -451,7 +451,8 @@ CAF_TEST(remote_actor_and_send) { ...@@ -451,7 +451,8 @@ CAF_TEST(remote_actor_and_send) {
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 mm = get_middleman_actor();
actor result; actor result;
auto f = self()->sync_send(mm, get_atom::value, "localhost", uint16_t{4242}); auto f = self()->sync_send(mm, connect_atom::value,
"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);
// build a fake server handshake containing the id of our first pseudo actor // build a fake server handshake containing the id of our first pseudo actor
...@@ -472,13 +473,15 @@ CAF_TEST(remote_actor_and_send) { ...@@ -472,13 +473,15 @@ CAF_TEST(remote_actor_and_send) {
invalid_actor_id, pseudo_remote(0)->id()}); invalid_actor_id, pseudo_remote(0)->id()});
// basp broker should've send the proxy // basp broker should've send the proxy
f.await( f.await(
[&](ok_atom, actor_addr res) { [&](ok_atom, node_id nid, actor_addr res, std::set<std::string> ifs) {
auto aptr = actor_cast<abstract_actor_ptr>(res); auto aptr = actor_cast<abstract_actor_ptr>(res);
CAF_REQUIRE(aptr.downcast<forwarding_actor_proxy>() != nullptr); CAF_REQUIRE(aptr.downcast<forwarding_actor_proxy>() != nullptr);
CAF_CHECK(get_namespace().get_all().size() == 1); CAF_CHECK(get_namespace().get_all().size() == 1);
CAF_CHECK(get_namespace().count_proxies(remote_node(0)) == 1); CAF_CHECK(get_namespace().count_proxies(remote_node(0)) == 1);
CAF_CHECK(nid == remote_node(0));
CAF_CHECK(res.node() == remote_node(0)); CAF_CHECK(res.node() == remote_node(0));
CAF_CHECK(res.id() == pseudo_remote(0)->id()); CAF_CHECK(res.id() == pseudo_remote(0)->id());
CAF_CHECK(ifs.empty());
auto proxy = get_namespace().get(remote_node(0), pseudo_remote(0)->id()); auto proxy = get_namespace().get(remote_node(0), pseudo_remote(0)->id());
CAF_REQUIRE(proxy != nullptr); CAF_REQUIRE(proxy != nullptr);
CAF_REQUIRE(proxy->address() == res); CAF_REQUIRE(proxy->address() == res);
......
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