Commit dd49a57d authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/remoting-overhaul' into develop

parents 456e7d04 3cb67a3c
No preview for this file type
doc/basp_header.png

39.3 KB | W: | H:

doc/basp_header.png

76.8 KB | W: | H:

doc/basp_header.png
doc/basp_header.png
doc/basp_header.png
doc/basp_header.png
  • 2-up
  • Swipe
  • Onion skin
......@@ -35,7 +35,6 @@ using plus_atom = atom_constant<atom("plus")>;
using minus_atom = atom_constant<atom("minus")>;
using result_atom = atom_constant<atom("result")>;
using rebind_atom = atom_constant<atom("rebind")>;
using connect_atom = atom_constant<atom("connect")>;
using reconnect_atom = atom_constant<atom("reconnect")>;
// our "service"
......@@ -105,9 +104,13 @@ private:
behavior reconnecting(std::function<void()> continuation = nullptr) {
using std::chrono::seconds;
auto mm = io::get_middleman_actor();
send(mm, get_atom::value, host_, port_);
send(mm, connect_atom::value, host_, port_);
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;
server_ = actor_cast<actor>(new_server);
// return to previous behavior
......@@ -122,7 +125,7 @@ private:
<< ": " << errstr
<< " [try again in 3s]"
<< 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) {
aout(this) << "*** rebind to " << nhost << ":" << nport << endl;
......@@ -131,13 +134,17 @@ private:
swap(port_, nport);
auto send_mm = [=] {
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
become(
keep_behavior,
[=](ok_atom&, actor_addr&) { send_mm(); },
[=](error_atom&, string&) { send_mm(); }
[=](ok_atom&, actor_addr&) {
send_mm();
},
[=](error_atom&, string&) {
send_mm();
}
);
},
// simply ignore all requests until we have a connection
......
......@@ -24,6 +24,7 @@ set (LIBCAF_CORE_SRCS
src/actor_proxy.cpp
src/actor_registry.cpp
src/attachable.cpp
src/announce_actor_type.cpp
src/behavior.cpp
src/behavior_stack.cpp
src/behavior_impl.cpp
......@@ -77,7 +78,8 @@ set (LIBCAF_CORE_SRCS
src/sync_request_bouncer.cpp
src/try_match.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp)
src/uniform_type_info_map.cpp
src/whereis.cpp)
add_custom_target(libcaf_core)
......
......@@ -156,6 +156,8 @@ protected:
public:
/// @cond PRIVATE
static actor_id latest_actor_id();
enum linking_operation {
establish_link_op,
establish_backlink_op,
......@@ -171,6 +173,8 @@ public:
static constexpr int is_blocking_flag = 0x10; // blocking_actor
static constexpr int is_detached_flag = 0x20; // 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) {
auto x = flags();
......@@ -227,6 +231,22 @@ public:
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`.
optional<uint32_t> handle(const std::exception_ptr& eptr);
......
......@@ -135,6 +135,11 @@ public:
/// Exchange content of `*this` and `other`.
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
inline abstract_actor* operator->() const noexcept {
......
......@@ -20,9 +20,9 @@
#ifndef CAF_ACTOR_NAMESPACE_HPP
#define CAF_ACTOR_NAMESPACE_HPP
#include <map>
#include <utility>
#include <functional>
#include <unordered_map>
#include "caf/node_id.hpp"
#include "caf/actor_cast.hpp"
......@@ -56,7 +56,7 @@ public:
/// Writes an actor address to `sink` and adds the actor
/// 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
/// addresses for remote actors on the fly if needed.
......@@ -120,7 +120,7 @@ public:
private:
backend& backend_;
std::map<key_type, proxy_map> proxies_;
std::unordered_map<key_type, proxy_map> proxies_;
};
} // namespace caf
......
......@@ -21,6 +21,7 @@
#define CAF_ATOM_HPP
#include <string>
#include <functional>
#include <type_traits>
#include "caf/detail/atom_val.hpp"
......@@ -68,6 +69,9 @@ using get_atom = atom_constant<atom("GET")>;
/// Generic 'PUT' atom for request operations.
using put_atom = atom_constant<atom("PUT")>;
/// Generic 'UPDATE' atom, e.g., or signalizing updates in a key-value store.
using update_atom = atom_constant<atom("UPDATE")>;
/// Generic 'DELETE' atom for request operations.
using delete_atom = atom_constant<atom("DELETE")>;
......@@ -104,6 +108,45 @@ using link_atom = atom_constant<atom("LINK")>;
/// Generic 'UNLINK' atom for removing networked links.
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 'PUBLISH' atom, e.g., for publishing actors at a given port.
using subscribe_atom = atom_constant<atom("SUBSCRIBE")>;
/// Generic 'UNPUBLISH' atom, e.g., for removing an actor/port mapping.
using unsubscribe_atom = atom_constant<atom("UNSUBSCRIB")>;
/// 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")>;
/// Generic 'SPAWN' atom, e.g., for spawning remote actors.
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 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
......@@ -46,6 +46,8 @@ class blocking_actor
: public extend<local_actor, blocking_actor>::
with<mixin::sync_sender<blocking_response_handle_tag>::impl> {
public:
using behavior_type = behavior;
blocking_actor();
~blocking_actor();
......
......@@ -127,6 +127,11 @@ operator>>(deserializer& source, T& value) {
return source.read(value, uniform_typeid<T>());
}
template <class T>
void operator&(deserializer& source, T& value) {
source >> value;
}
} // namespace caf
#endif // CAF_DESERIALIZER_HPP
......@@ -20,13 +20,14 @@
#ifndef CAF_DETAIL_ACTOR_REGISTRY_HPP
#define CAF_DETAIL_ACTOR_REGISTRY_HPP
#include <map>
#include <mutex>
#include <thread>
#include <atomic>
#include <cstdint>
#include <unordered_map>
#include <condition_variable>
#include "caf/actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/abstract_actor.hpp"
......@@ -38,10 +39,8 @@ namespace detail {
class singletons;
class actor_registry : public singleton_mixin<actor_registry> {
class actor_registry {
public:
friend class singleton_mixin<actor_registry>;
~actor_registry();
/// A registry entry consists of a pointer to the actor and an
......@@ -82,19 +81,37 @@ public:
// blocks the caller until running-actors-count becomes `expected`
void await_running_count_equal(size_t expected);
actor get_named(atom_value key) const;
void put_named(atom_value key, actor value);
using named_entries = std::unordered_map<atom_value, actor>;
named_entries named_actors() const;
static actor_registry* create_singleton();
void dispose();
void stop();
void initialize();
private:
using entries = std::map<actor_id, value_type>;
using entries = std::unordered_map<actor_id, value_type>;
actor_registry();
std::atomic<size_t> running_;
std::atomic<actor_id> ids_;
std::mutex running_mtx_;
std::condition_variable running_cv_;
mutable detail::shared_spinlock instances_mtx_;
entries entries_;
named_entries named_entries_;
mutable detail::shared_spinlock named_entries_mtx_;
};
} // namespace detail
......
......@@ -493,6 +493,26 @@ public:
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 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. *
******************************************************************************/
#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/infer_handle.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
namespace experimental {
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_class<T>::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
......@@ -29,6 +29,18 @@
namespace caf {
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
/// before `make_behavior` will get called and destroyed after the
/// actor called `quit`. This state management brakes cycles and
......@@ -39,7 +51,8 @@ class stateful_actor : public Base {
public:
template <class... Ts>
stateful_actor(Ts&&... xs) : Base(std::forward<Ts>(xs)...), state(state_) {
// nop
if (detail::is_serializable<State>::value)
this->is_serializable(true);
}
~stateful_actor() {
......@@ -55,6 +68,14 @@ public:
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.
State& state;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_WHEREIS_HPP
#define CAF_EXPERIMENTAL_WHEREIS_HPP
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
namespace caf {
namespace experimental {
actor whereis(atom_value registered_name);
} // namespace experimental
} // namespace caf
#endif // CAF_EXPERIMENTAL_WHEREIS_HPP
......@@ -56,6 +56,8 @@ class mailbox_element;
class message_handler;
class uniform_type_info;
class event_based_actor;
class binary_serializer;
class binary_deserializer;
class forwarding_actor_proxy;
// structs
......@@ -87,8 +89,16 @@ template <class T, typename U>
T actor_cast(const U&);
namespace io {
class broker;
class middleman;
class broker;
class middleman;
namespace basp {
struct header;
} // namespace basp
} // namespace io
namespace scheduler {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_INFER_HANDLE_HPP
#define CAF_INFER_HANDLE_HPP
#include "caf/actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_event_based_actor.hpp"
namespace caf {
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_impl {
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_impl<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_impl<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_impl<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_impl<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_impl<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...>;
};
template <class T>
struct infer_handle_from_class {
using type =
typename infer_handle_from_behavior<
typename T::behavior_type
>::type;
static constexpr spawn_mode mode = spawn_mode::clazz;
};
} // namespace caf
#endif // CAF_INFER_HANDLE_HPP
......@@ -392,6 +392,16 @@ public:
/// implementation simply returns "actor".
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 *
****************************************************************************/
......@@ -463,7 +473,9 @@ public:
return (mid.is_request()) ? mid.response_id() : message_id();
}
void forward_message(const actor& dest, message_priority mp);
void forward_current_message(const actor& dest);
void forward_current_message(const actor& dest, message_priority mp);
template <class... Ts>
void delegate(message_priority mp, const actor& dest, Ts&&... xs) {
......@@ -611,9 +623,9 @@ public:
initial_behavior_fac_ = std::move(fun);
}
protected:
void do_become(behavior bhvr, bool discard_old);
protected:
// used only in thread-mapped actors
void await_data();
......
......@@ -390,6 +390,12 @@ inline message make_message(message other) {
return std::move(other);
}
/// Returns an empty `message`.
/// @relates message
inline message make_message() {
return message{};
}
/******************************************************************************
* template member function implementations *
******************************************************************************/
......
......@@ -49,9 +49,13 @@ public:
}
/// Sends `response_message` and invalidates this handle afterwards.
void deliver(message response_message) const;
template <class... Ts>
void deliver(Ts&&... xs) const {
deliver_impl(make_message(std::forward<Ts>(xs)...));
}
private:
void deliver_impl(message response_message) const;
actor_addr from_;
actor_addr to_;
message_id id_;
......
......@@ -107,6 +107,11 @@ operator<<(serializer& sink, const T& value) {
return sink.write(value, uniform_typeid<T>());
}
template <class T>
void operator&(serializer& sink, const T& value) {
sink << value;
}
} // namespace caf
#endif // CAF_SERIALIZER_HPP
......@@ -222,7 +222,7 @@ struct infer_typed_actor_base<Result, T*, true> {
/// constructor arguments.
template <class C, spawn_options Os = no_spawn_options, class... Ts>
typename actor_handle_from_signature_list<typename C::signatures>::type
spawn_typed(Ts&&... xs) {
CAF_DEPRECATED spawn_typed(Ts&&... xs) {
return spawn_class<C, Os>(nullptr, empty_before_launch_callback{},
std::forward<Ts>(xs)...);
}
......
......@@ -228,6 +228,12 @@ private:
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
#endif // CAF_TYPED_BEHAVIOR_HPP
......@@ -41,9 +41,16 @@
namespace caf {
namespace {
using guard_type = std::unique_lock<std::mutex>;
std::atomic<actor_id> ids_;
} // namespace <anonymous>
actor_id abstract_actor::latest_actor_id() {
return ids_.load();
}
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
......@@ -59,7 +66,7 @@ abstract_actor::abstract_actor(actor_id aid, node_id nid)
abstract_actor::abstract_actor()
: abstract_channel(abstract_channel::is_abstract_actor_flag,
detail::singletons::get_node_id()),
id_(detail::singletons::get_actor_registry()->next_id()),
id_(++ids_),
exit_reason_(exit_reason::not_exited),
host_(nullptr) {
// nop
......
......@@ -62,7 +62,7 @@ actor_namespace::actor_namespace(backend& be) : backend_(be) {
// 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);
if (! addr) {
node_id::host_id_type zero;
......
......@@ -22,11 +22,19 @@
#include <mutex>
#include <limits>
#include <stdexcept>
#include <unordered_map>
#include <unordered_set>
#include "caf/spawn.hpp"
#include "caf/locks.hpp"
#include "caf/actor_cast.hpp"
#include "caf/attachable.hpp"
#include "caf/exit_reason.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/experimental/stateful_actor.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#include "caf/scheduler/detached_threads.hpp"
......@@ -47,7 +55,7 @@ actor_registry::~actor_registry() {
// nop
}
actor_registry::actor_registry() : running_(0), ids_(1) {
actor_registry::actor_registry() : running_(0) {
// nop
}
......@@ -95,10 +103,6 @@ void actor_registry::erase(actor_id key, uint32_t reason) {
}
}
uint32_t actor_registry::next_id() {
return ++ids_;
}
void actor_registry::inc_running() {
# if defined(CAF_LOG_LEVEL) && CAF_LOG_LEVEL >= CAF_DEBUG
CAF_LOG_DEBUG("new value = " << ++running_);
......@@ -130,5 +134,119 @@ void actor_registry::await_running_count_equal(size_t expected) {
}
}
actor actor_registry::get_named(atom_value key) const {
shared_guard guard{named_entries_mtx_};
auto i = named_entries_.find(key);
if (i == named_entries_.end())
return invalid_actor;
return i->second;
}
void actor_registry::put_named(atom_value key, actor value) {
if (value)
value->attach_functor([=](uint32_t) {
detail::singletons::get_actor_registry()->put_named(key, invalid_actor);
});
exclusive_guard guard{named_entries_mtx_};
named_entries_.emplace(key, std::move(value));
}
auto actor_registry::named_actors() const -> named_entries {
shared_guard guard{named_entries_mtx_};
return named_entries_;
}
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() {
using namespace experimental;
struct kvstate {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<actor>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<actor,topic_set> subscribers;
const char* name = "caf.config_server";
};
auto kvstore = [](stateful_actor<kvstate>* self) -> behavior {
return {
[=](put_atom, const std::string& key, message& msg) {
auto& vp = self->state.data[key];
if (vp.first == msg)
return;
vp.first = std::move(msg);
for (auto& subscriber : vp.second)
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
},
[=](get_atom, std::string& key) -> message {
auto i = self->state.data.find(key);
return make_message(ok_atom::value, std::move(key),
i != self->state.data.end() ? i->second.first
: make_message());
},
[=](subscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
if (! subscriber)
return;
self->state.data[key].second.insert(subscriber);
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i != subscribers.end()) {
i->second.insert(key);
} else {
self->monitor(subscriber);
subscribers.emplace(subscriber, kvstate::topic_set{key});
}
},
[=](unsubscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
if (! subscriber)
return;
self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber);
},
[=](const down_msg& dm) {
auto subscriber = actor_cast<actor>(dm.source);
if (! subscriber)
return;
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(subscriber);
subscribers.erase(i);
},
others >> [] {
return make_message(error_atom::value, "unsupported operation");
}
};
};
named_entries_.emplace(atom("SpawnServ"), spawn_announce_actor_type_server());
named_entries_.emplace(atom("ConfigServ"), spawn<hidden>(kvstore));
}
} // namespace detail
} // 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)
-> either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string> {
auto i = self->state.funs_.find(name);
if (i == self->state.funs_.end())
return {error_atom::value, "no actor type found named " + name};
auto f = i->second;
auto res = f(args);
if (res.first == invalid_actor_addr)
return {error_atom::value, "cannot initialize an actor type " + name
+ " using the provided arguments"};
return {ok_atom::value, res.first, res.second};
},
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("SpawnServ"));
anon_send(server, add_atom::value, std::move(name), std::move(f));
}
} // namespace experimental
} // namespace caf
......@@ -30,7 +30,7 @@ event_based_actor::~event_based_actor() {
void event_based_actor::forward_to(const actor& whom,
message_priority prio) {
forward_message(whom, prio);
forward_current_message(whom, prio);
}
void event_based_actor::initialize() {
......
......@@ -108,10 +108,16 @@ std::vector<group> local_actor::joined_groups() const {
return result;
}
void local_actor::forward_message(const actor& dest, message_priority prio) {
if (! dest) {
void local_actor::forward_current_message(const actor& dest) {
if (! dest)
return;
dest->enqueue(std::move(current_element_), host());
}
void local_actor::forward_current_message(const actor& dest,
message_priority prio) {
if (! dest)
return;
}
auto mid = current_element_->mid;
current_element_->mid = prio == message_priority::high
? mid.with_high_priority()
......@@ -180,27 +186,117 @@ enum class msg_type {
expired_sync_response, // a sync response that already timed out
timeout, // triggers currently active timeout
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) {
const message& msg = node.msg;
message& msg = node.msg;
auto mid = node.mid;
if (mid.is_response()) {
if (mid.is_response())
return self->awaits(mid) ? msg_type::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;
}
if (msg.size() != 1) {
return msg_type::ordinary;
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());
},
[&](sys_atom, get_atom, std::string& what) {
CAF_LOGF_TRACE(CAF_ARG(what));
if (what == "info") {
CAF_LOGF_DEBUG("reply to 'info' message");
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
ok_atom::value, std::move(what),
self->address(), self->name()),
self->host());
return;
}
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
error_atom::value,
"unknown key: " + std::move(what)),
self->host());
},
others >> [&] {
mismatch = true;
}
});
return mismatch ? msg_type::ordinary : msg_type::sys_message;
}
// all other system messages always consist of one element
if (msg.size() != 1)
return msg_type::ordinary;
if (msg.match_element<timeout_msg>(0)) {
auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(! mid.valid());
if (self->is_active_timeout(tid)) {
return msg_type::timeout;
}
return msg_type::expired_timeout;
return self->is_active_timeout(tid) ? msg_type::timeout
: msg_type::expired_timeout;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
......@@ -316,6 +412,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
case msg_type::sys_message:
CAF_LOG_DEBUG("handled system message");
return im_dropped;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
......@@ -348,7 +447,6 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
} else {
auto res = post_process_invoke_res(this, mid,
fun(current_mailbox_element()->msg));
ptr.swap(current_mailbox_element());
if (! res) {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << id());
......@@ -824,6 +922,14 @@ const char* local_actor::name() const {
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() {
return pending_responses_.empty() ? bhvr_stack_.back()
: pending_responses_.front().second;
......
......@@ -101,8 +101,9 @@ const char* message::uniform_name_at(size_t pos) const {
}
bool message::equals(const message& other) const {
CAF_ASSERT(vals_);
return vals_->equals(*other.vals());
if (empty())
return other.empty();
return other.empty() ? false : vals_->equals(*other.vals());
}
message message::drop(size_t n) const {
......
......@@ -30,10 +30,9 @@ response_promise::response_promise(const actor_addr& from, const actor_addr& to,
CAF_ASSERT(id.is_response() || ! id.valid());
}
void response_promise::deliver(message msg) const {
if (! to_) {
void response_promise::deliver_impl(message msg) const {
if (! to_)
return;
}
auto to = actor_cast<abstract_actor_ptr>(to_);
auto from = actor_cast<abstract_actor_ptr>(from_);
to->enqueue(from_, id_, std::move(msg), from->host());
......
......@@ -44,6 +44,11 @@ std::thread run_program_impl(actor rc, const char* cpath,
oss << " 2>&1";
string cmdstr = oss.str();
return std::thread{ [cmdstr, rc] {
// on FreeBSD, popen() hangs indefinitely in some cases
# ifdef CAF_BSD
system(cmdstr.c_str());
anon_send(rc, "");
# else
string output;
auto fp = popen(cmdstr.c_str(), "r");
if (! fp) {
......@@ -57,6 +62,7 @@ std::thread run_program_impl(actor rc, const char* cpath,
}
pclose(fp);
anon_send(rc, output);
# endif
}};
}
#else
......
......@@ -76,12 +76,12 @@ void singletons::stop_singletons() {
}
CAF_LOGF_DEBUG("stop group manager");
stop(s_group_manager);
CAF_LOGF_DEBUG("stop actor registry");
stop(s_actor_registry);
CAF_LOGF_DEBUG("stop scheduler");
stop(s_scheduling_coordinator);
CAF_LOGF_DEBUG("wait for all detached threads");
scheduler::await_detached_threads();
CAF_LOGF_DEBUG("stop actor registry");
stop(s_actor_registry);
// dispose singletons, i.e., release memory
CAF_LOGF_DEBUG("dispose plugins");
for (auto& plugin : s_plugins) {
......
......@@ -95,18 +95,6 @@ const char* numbered_type_names[] = {
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
template <class T, class U>
typename std::enable_if<
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/whereis.hpp"
#include "caf/actor.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
namespace experimental {
actor whereis(atom_value registered_name) {
return detail::singletons::get_actor_registry()->get_named(registered_name);
}
} // namespace experimental
} // 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/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("SpawnServ"));
}
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();
}
};
} // 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()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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()
......@@ -30,6 +30,67 @@ using std::endl;
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 {
~fixture() {
await_all_actors_done();
......@@ -37,6 +98,7 @@ struct fixture {
}
};
constexpr const char* global_redirect = ":test";
constexpr const char* local_redirect = ":test2";
......@@ -57,6 +119,133 @@ void chattier_actor(event_based_actor* self, const std::string& fn) {
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) {
scoped_actor self;
self->join(group::get("local", global_redirect));
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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();
}
......@@ -27,6 +27,18 @@
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) {
auto m1 = make_message(1, 2, 3, 4, 5);
std::vector<message> messages{
......
......@@ -32,6 +32,7 @@
#include "caf/io/remote_group.hpp"
#include "caf/io/set_middleman.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/middleman_actor.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/publish_local_groups.hpp"
......
......@@ -27,14 +27,16 @@
#include <unordered_map>
#include <unordered_set>
#include "caf/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/callback.hpp"
#include "caf/actor_addr.hpp"
#include "caf/serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/callback.hpp"
#include "caf/actor_namespace.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/system_messages.hpp"
......@@ -307,6 +309,11 @@ public:
/// `invalid_connection_handle` if no direct connection to `nid` exists.
connection_handle lookup_direct(const node_id& nid) const;
/// Returns the next hop that would be chosen for `nid`
/// or `invalid_node_id` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const;
/// Flush output buffer for `r`.
void flush(const route& r);
......@@ -315,7 +322,7 @@ public:
void add_direct(const connection_handle& hdl, const node_id& dest);
/// Adds a new indirect route to the table.
void add_indirect(const node_id& hop, const node_id& dest);
bool add_indirect(const node_id& hop, const node_id& dest);
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
......@@ -325,6 +332,10 @@ public:
/// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`.
bool erase_indirect(const node_id& dest);
/// Queries whether `dest` is reachable.
bool reachable(const node_id& dest);
......@@ -362,14 +373,14 @@ public:
/// Provides a callback-based interface for certain BASP events.
class callee {
public:
explicit callee(actor_namespace::backend& mgm);
explicit callee(actor_namespace::backend& mgm, middleman& mm);
virtual ~callee();
/// Called if a server handshake was received and
/// the connection to `nid` is established.
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
/// node became unrechable for other reasons *before*
......@@ -378,8 +389,12 @@ public:
/// routing table from this callback.
virtual void purge_state(const node_id& nid) = 0;
/// Called whenever a remote node created a proxy
/// for one of our local actors.
virtual void proxy_announced(const node_id& nid, actor_id aid) = 0;
/// Called whenever a remote actor died to destroy
/// the proxy instance on our end.
virtual void kill_proxy(const node_id& nid, actor_id aid, uint32_t rsn) = 0;
/// Called whenever a `dispatch_message` arrived for a local actor.
......@@ -387,13 +402,27 @@ public:
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Returns the actor namespace associated to this BASP protocol instance.
actor_namespace& get_namespace() {
inline actor_namespace& get_namespace() {
return namespace_;
}
public:
inline middleman& get_middleman() {
return middleman_;
}
protected:
actor_namespace namespace_;
middleman& middleman_;
};
/// Describes a function object responsible for writing
......@@ -490,6 +519,9 @@ public:
/// written (e.g. used when establishing direct connections on-the-fly).
void write_server_handshake(buffer_type& buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`.
void write_client_handshake(buffer_type& buf, const node_id& remote_side);
/// Writes a `dispatch_error` to `buf`.
void write_dispatch_error(buffer_type& buf,
const node_id& source_node,
......@@ -504,10 +536,16 @@ public:
actor_id aid,
uint32_t rsn);
const node_id& this_node() const {
inline const node_id& this_node() const {
return this_node_;
}
/// Invokes the callback(s) associated with given event.
template <hook::event_type Event, typename... Ts>
void notify(Ts&&... xs) {
callee_.get_middleman().template notify<Event>(std::forward<Ts>(xs)...);
}
private:
routing_table tbl_;
published_actor_map published_actors_;
......
......@@ -49,7 +49,7 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// inherited from basp::instance::listener
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
void purge_state(const node_id& id) override;
......@@ -60,10 +60,21 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// inherited from basp::instance::listener
void kill_proxy(const node_id& nid, actor_id aid, uint32_t rsn) override;
// inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) override;
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
// inherited from basp::instance::listener
void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly_before) override;
// inherited from basp::instance::listener
void learned_new_node_indirectly(const node_id& nid) override;
struct connection_context {
basp::connection_state cstate;
basp::header hdr;
......@@ -71,7 +82,6 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
node_id id;
uint16_t remote_port;
optional<response_promise> callback;
std::set<std::string> expected_sigs;
};
void set_context(connection_handle hdl);
......@@ -94,6 +104,16 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// keeps the associated proxies alive to work around subtle bugs
std::unordered_map<node_id, std::pair<uint16_t, actor_addr>> known_remotes;
// stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to
// get a 'SpawnServ' instance on the remote side
std::unordered_map<node_id, actor> spawn_servers;
// can be enabled by the user to let CAF automatically try
// to establish new connections at runtime to optimize
// routing paths by forming a mesh between all nodes
bool enable_automatic_connections = false;
const node_id& this_node() const {
return instance.this_node();
}
......
......@@ -20,6 +20,7 @@
#ifndef CAF_IO_HOOK_HPP
#define CAF_IO_HOOK_HPP
#include <set>
#include <memory>
#include <vector>
......@@ -71,16 +72,17 @@ public:
const message& payload);
/// Called whenever a message is forwarded to a different node.
virtual void message_forwarded_cb(const node_id& from, const node_id& dest,
virtual void message_forwarded_cb(const basp::header& hdr,
const std::vector<char>* payload);
/// Called whenever no route for a forwarding request exists.
virtual void message_forwarding_failed_cb(const node_id& from,
const node_id& to,
virtual void message_forwarding_failed_cb(const basp::header& hdr,
const std::vector<char>* payload);
/// Called whenever an actor has been published.
virtual void actor_published_cb(const actor_addr& addr, uint16_t port);
virtual void actor_published_cb(const actor_addr& addr,
const std::set<std::string>& ifs,
uint16_t port);
/// Called whenever a new remote actor appeared.
virtual void new_remote_actor_cb(const actor_addr& addr);
......@@ -93,6 +95,15 @@ public:
/// @param node The newly added entry to the routing table.
virtual void new_route_added_cb(const node_id& via, const node_id& node);
/// Called whenever a direct connection was lost.
virtual void connection_lost_cb(const node_id& dest);
/// Called whenever a route became unavailable.
/// @param hop The node that was either disconnected
/// or lost a connection itself.
/// @param dest The node that is no longer reachable via `hop`.
virtual void route_lost_cb(const node_id& hop, const node_id& dest);
/// Called whenever a message was discarded because a remote node
/// tried to send a message to an actor ID that could not be found
/// in the registry.
......@@ -115,6 +126,8 @@ public:
new_remote_actor,
new_connection_established,
new_route_added,
connection_lost,
route_lost,
invalid_message_received,
before_shutdown
};
......@@ -150,6 +163,8 @@ private:
CAF_IO_HOOK_DISPATCH(new_remote_actor)
CAF_IO_HOOK_DISPATCH(new_connection_established)
CAF_IO_HOOK_DISPATCH(new_route_added)
CAF_IO_HOOK_DISPATCH(connection_lost)
CAF_IO_HOOK_DISPATCH(route_lost)
CAF_IO_HOOK_DISPATCH(invalid_message_received)
CAF_IO_HOOK_DISPATCH(before_shutdown)
};
......
......@@ -99,6 +99,10 @@ public:
});
}
inline bool has_hook() const {
return hooks_ != nullptr;
}
template <class F>
void add_shutdown_cb(F fun) {
struct impl : hook {
......
......@@ -30,100 +30,94 @@ namespace io {
///
/// 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 {
/// (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)
/// -> put_result;
/// // 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.
/// // @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)
/// -> put_result;
/// // Opens a new port other CAF instances can connect to. The
/// // 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)
/// -> put_result;
/// // Queries a remote node and returns an ID to this node as well as
/// // 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)
/// -> get_result;
/// // Closes `port` if it is mapped to `whom`.
/// // @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)
/// -> get_result;
/// // Unconditionally closes `port`, removing any actor
/// // 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;
/// // Spawns an actor on a remote node, initializing it
/// // using the arguments stored in `msg`.
/// // @param nid ID of the remote node that should spawn the actor.
/// // @param name Announced type name of the actor.
/// // @param args Initialization arguments for the actor.
/// // @returns The address of the spawned actor and its interface
/// // description on success; an error string otherwise.
/// (spawn_atom, node_id nid, string name, message args)
/// -> either (ok_atom, actor_addr, set<string>
/// or (error_atom, string error_string)
///
/// (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 =
typed_actor<
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>, std::string, bool>
::with_either<ok_atom, uint16_t>
::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>
::or_else<error_atom, std::string>,
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>, bool>
replies_to<publish_atom, uint16_t, actor_addr,
std::set<std::string>, std::string, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, uint16_t, actor_addr, std::set<std::string>>
replies_to<open_atom, uint16_t, std::string, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t, std::set<std::string>>
::with_either<ok_atom, actor_addr>
replies_to<connect_atom, std::string, uint16_t>
::with_either<ok_atom, node_id, actor_addr, std::set<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>
::or_else<error_atom, std::string>,
replies_to<delete_atom, actor_addr, uint16_t>
replies_to<close_atom, uint16_t>
::with_either<ok_atom>
::or_else<error_atom, std::string>,
replies_to<spawn_atom, node_id, std::string, message>
::with_either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string>>;
/// Returns a handle for asynchronous networking operations.
......@@ -133,4 +127,3 @@ middleman_actor get_middleman_actor(); // implemented in middleman.cpp
} // namespace caf
#endif // CAF_IO_MIDDLEMAN_ACTOR_HPP
......@@ -33,10 +33,11 @@ namespace caf {
namespace io {
namespace network {
// {protocol => address}
using address_listing = std::map<protocol, std::vector<std::string>>;
// {interface_name => {protocol => address}}
using interfaces_map = std::map<std::string,
std::map<protocol,
std::vector<std::string>>>;
using interfaces_map = std::map<std::string, address_listing>;
/// Utility class bundling access to network interface names and addresses.
class interfaces {
......@@ -45,8 +46,7 @@ public:
static interfaces_map list_all(bool include_localhost = true);
/// Returns all addresses for all devices for all protocols.
static std::map<protocol, std::vector<std::string>>
list_addresses(bool include_localhost = true);
static address_listing list_addresses(bool include_localhost = true);
/// Returns all addresses for all devices for given protocol.
static std::vector<std::string> list_addresses(protocol proc,
......
......@@ -31,8 +31,8 @@
namespace caf {
namespace io {
abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port);
actor_addr remote_actor_impl(std::set<std::string> ifs,
std::string host, uint16_t port);
/// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
......@@ -41,8 +41,8 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
/// representing a remote actor.
/// @throws network_error Thrown on connection error or
/// when connecting to a typed actor.
inline actor remote_actor(const std::string& host, uint16_t port) {
auto res = remote_actor_impl(std::set<std::string>{}, host, port);
inline actor remote_actor(std::string host, uint16_t port) {
auto res = remote_actor_impl(std::set<std::string>{}, std::move(host), port);
return actor_cast<actor>(res);
}
......@@ -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
/// to an untyped otherwise unexpected actor.
template <class ActorHandle>
ActorHandle typed_remote_actor(const std::string& host, uint16_t port) {
auto iface = ActorHandle::message_types();
return actor_cast<ActorHandle>(remote_actor_impl(std::move(iface),
host, port));
ActorHandle typed_remote_actor(std::string host, uint16_t port) {
auto res = remote_actor_impl(ActorHandle::message_types(),
std::move(host), port);
return actor_cast<ActorHandle>(res);
}
} // namespace io
......
This diff is collapsed.
This diff is collapsed.
......@@ -40,9 +40,14 @@ void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node,
call_next<message_sent>(from, dest_node, dest, mid, payload);
}
void hook::message_forwarded_cb(const node_id& from, const node_id& dest,
void hook::message_forwarded_cb(const basp::header& hdr,
const std::vector<char>* payload) {
call_next<message_forwarded>(from, dest, payload);
call_next<message_forwarded>(hdr, payload);
}
void hook::message_forwarding_failed_cb(const basp::header& hdr,
const std::vector<char>* payload) {
call_next<message_forwarding_failed>(hdr, payload);
}
void hook::message_sending_failed_cb(const actor_addr& from,
......@@ -52,13 +57,10 @@ void hook::message_sending_failed_cb(const actor_addr& from,
call_next<message_sending_failed>(from, dest, mid, payload);
}
void hook::message_forwarding_failed_cb(const node_id& from, const node_id& to,
const std::vector<char>* payload) {
call_next<message_forwarding_failed>(from, to, payload);
}
void hook::actor_published_cb(const actor_addr& addr, uint16_t port) {
call_next<actor_published>(addr, port);
void hook::actor_published_cb(const actor_addr& addr,
const std::set<std::string>& ifs,
uint16_t port) {
call_next<actor_published>(addr, ifs, port);
}
void hook::new_remote_actor_cb(const actor_addr& addr) {
......@@ -73,6 +75,14 @@ void hook::new_route_added_cb(const node_id& via, const node_id& node) {
call_next<new_route_added>(via, node);
}
void hook::connection_lost_cb(const node_id& dest) {
call_next<connection_lost>(dest);
}
void hook::route_lost_cb(const node_id& hop, const node_id& dest) {
call_next<route_lost>(hop, dest);
}
void hook::invalid_message_received_cb(const node_id& source,
const actor_addr& sender,
actor_id invalid_dest,
......
This diff is collapsed.
......@@ -53,7 +53,7 @@ uint16_t publish_impl(uint16_t port, actor_addr whom,
uint16_t result;
std::string error_msg;
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(
[&](ok_atom, uint16_t res) {
result = res;
......
......@@ -42,14 +42,20 @@
namespace caf {
namespace io {
abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port) {
actor_addr remote_actor_impl(std::set<std::string> ifs,
std::string host, uint16_t port) {
auto mm = get_middleman_actor();
actor_addr result;
scoped_actor self;
abstract_actor_ptr result;
self->sync_send(mm, get_atom{}, std::move(host), port, std::move(ifs)).await(
[&](ok_atom, actor_addr res) {
result = actor_cast<abstract_actor_ptr>(res);
self->sync_send(mm, connect_atom::value, std::move(host), port).await(
[&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) {
if (!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) {
throw network_error(std::move(msg));
......@@ -60,4 +66,3 @@ abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
} // namespace io
} // namespace caf
......@@ -61,9 +61,11 @@ void scribe::consume(const void*, size_t num_bytes) {
buf.resize(num_bytes);
read_msg().buf.swap(buf);
parent()->invoke_message(invalid_actor_addr, invalid_message_id, read_msg_);
if (! read_msg_.empty()) {
// swap buffer back to stream and implicitly flush wr_buf()
read_msg().buf.swap(buf);
flush();
}
}
void scribe::io_failure(network::operation op) {
......
......@@ -34,7 +34,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) {
auto mm = get_middleman_actor();
if (blocking) {
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, basp_broker is done
},
......@@ -43,7 +43,7 @@ void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) {
}
);
} else {
anon_send(mm, delete_atom::value, whom, port);
anon_send(mm, unpublish_atom::value, whom, port);
}
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 io_automatic_connection
#include "caf/test/unit_test.hpp"
#include <set>
#include <thread>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/experimental/whereis.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/detail/run_program.hpp"
using namespace caf;
using namespace caf::io;
using namespace caf::experimental;
using std::string;
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
/*
This test checks whether automatic connections work as expected
by first connecting three nodes "in line". In step 2, we send a
message across the line, forcing the nodes to build a mesh. In step 3,
we disconnect the node that originally connected the other two and expect
that the other two nodes communicate uninterrupted.
1) Initial setup:
Earth ---- Mars ---- Jupiter
2) After Jupiter has send a message to Earth:
Earth ---- Mars
\ /
\ /
\ /
Jupiter
3) After Earth has received the message and disconnected Mars:
Earth ---- Jupiter
*/
std::thread run_prog(const char* arg, uint16_t port, bool use_asio) {
return detail::run_program(invalid_actor, caf::test::engine::path(), "-n",
"-s", CAF_XSTR(CAF_SUITE), "--", arg, "-p", port,
(use_asio ? "--use-asio" : ""));
}
// we run the same code on all three nodes, a simple ping-pong client
struct testee_state {
std::set<actor> buddies;
uint16_t port = 0;
const char* name = "testee";
};
behavior testee(stateful_actor<testee_state>* self) {
return {
[self](ping_atom, actor buddy, bool please_broadcast) -> message {
if (please_broadcast)
for (auto& x : self->state.buddies)
if (x != buddy)
send_as(buddy, x, ping_atom::value, buddy, false);
self->state.buddies.emplace(std::move(buddy));
return make_message(pong_atom::value, self);
},
[self](pong_atom, actor buddy) {
self->state.buddies.emplace(std::move(buddy));
},
[self](put_atom, uint16_t new_port) {
self->state.port = new_port;
},
[self](get_atom) {
return self->state.port;
}
};
}
void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
scoped_actor self;
struct captain : hook {
public:
captain(actor parent) : parent_(std::move(parent)) {
// nop
}
void new_connection_established_cb(const node_id& node) override {
anon_send(parent_, put_atom::value, node);
call_next<hook::new_connection_established>(node);
}
void new_remote_actor_cb(const actor_addr& addr) override {
anon_send(parent_, put_atom::value, addr);
call_next<hook::new_remote_actor>(addr);
}
void connection_lost_cb(const node_id& dest) override {
anon_send(parent_, delete_atom::value, dest);
}
private:
actor parent_;
};
middleman::instance()->add_hook<captain>(self);
auto aut = spawn(testee);
auto port = publish(aut, pub_port);
CAF_TEST_VERBOSE("published testee at port " << port);
std::thread mars_process;
std::thread jupiter_process;
// launch process for Mars
if (! as_server) {
CAF_TEST_VERBOSE("launch process for Mars");
mars_process = run_prog("--mars", port, use_asio);
}
CAF_TEST_VERBOSE("wait for Mars to connect");
node_id mars;
self->receive(
[&](put_atom, const node_id& nid) {
mars = nid;
CAF_TEST_VERBOSE(CAF_TSARG(mars));
}
);
actor_addr mars_addr;
uint16_t mars_port;
self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
mars_addr = addr;
CAF_TEST_VERBOSE(CAF_TSARG(mars_addr));
self->sync_send(actor_cast<actor>(mars_addr), get_atom::value).await(
[&](uint16_t mp) {
CAF_TEST_VERBOSE("mars published its actor at port " << mp);
mars_port = mp;
}
);
}
);
}
);
// launch process for Jupiter
if (! as_server) {
CAF_TEST_VERBOSE("launch process for Jupiter");
jupiter_process = run_prog("--jupiter", mars_port, use_asio);
}
CAF_TEST_VERBOSE("wait for Jupiter to connect");
self->receive(
[](put_atom, const node_id& jupiter) {
CAF_TEST_VERBOSE(CAF_TSARG(jupiter));
}
);
actor_addr jupiter_addr;
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
jupiter_addr = addr;
CAF_TEST_VERBOSE(CAF_TSARG(jupiter_addr));
}
);
}
);
CAF_TEST_VERBOSE("shutdown Mars");
anon_send_exit(mars_addr, exit_reason::kill);
if (mars_process.joinable())
mars_process.join();
self->receive(
[&](delete_atom, const node_id& nid) {
CAF_CHECK_EQUAL(nid, mars);
}
);
CAF_TEST_VERBOSE("check whether we still can talk to Jupiter");
self->send(aut, ping_atom::value, self, true);
std::set<actor_addr> found;
int i = 0;
self->receive_for(i, 2)(
[&](pong_atom, const actor&) {
found.emplace(self->current_sender());
}
);
std::set<actor_addr> expected{aut.address(), jupiter_addr};
CAF_CHECK_EQUAL(found, expected);
CAF_TEST_VERBOSE("shutdown Jupiter");
anon_send_exit(jupiter_addr, exit_reason::kill);
if (jupiter_process.joinable())
jupiter_process.join();
anon_send_exit(aut, exit_reason::kill);
}
void run_mars(uint16_t port_to_earth, uint16_t pub_port) {
auto aut = spawn(testee);
auto port = publish(aut, pub_port);
anon_send(aut, put_atom::value, port);
CAF_TEST_VERBOSE("published testee at port " << port);
auto earth = remote_actor("localhost", port_to_earth);
send_as(aut, earth, ping_atom::value, aut, false);
}
void run_jupiter(uint16_t port_to_mars) {
auto aut = spawn(testee);
auto mars = remote_actor("localhost", port_to_mars);
send_as(aut, mars, ping_atom::value, aut, true);
}
CAF_TEST(triangle_setup) {
uint16_t port = 0;
uint16_t publish_port = 0;
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
auto r = message_builder(argv, argv + argc).extract_opts({
{"port,p", "port of remote side (when running mars or jupiter)", port},
{"mars", "run mars"},
{"jupiter", "run jupiter"},
{"use-asio", "use ASIO network backend (if available)"},
{"server,s", "run in server mode (don't run clients)", publish_port}
});
// check arguments
bool is_mars = r.opts.count("mars") > 0;
bool is_jupiter = r.opts.count("jupiter") > 0;
bool has_port = r.opts.count("port") > 0;
if (((is_mars || is_jupiter) && ! has_port) || (is_mars && is_jupiter)) {
CAF_TEST_ERROR("need a port when running Mars or Jupiter and cannot "
"both at the same time");
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
CAF_MESSAGE("enable ASIO backend");
set_middleman<network::asio_multiplexer>();
}
# endif // CAF_USE_ASIO
// enable automatic connections
anon_send(whereis(atom("ConfigServ")), put_atom::value,
"global.enable-automatic-connections", make_message(true));
auto as_server = r.opts.count("server") > 0;
if (is_mars)
run_mars(port, publish_port);
else if (is_jupiter)
run_jupiter(port);
else
run_earth(use_asio, as_server, publish_port);
await_all_actors_done();
shutdown();
}
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 io_remote_spawn
#include "caf/test/unit_test.hpp"
#include <thread>
#include <string>
#include <cstring>
#include <sstream>
#include <iostream>
#include <functional>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/run_program.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf;
using namespace caf::experimental;
namespace {
behavior mirror(event_based_actor* self) {
return {
others >> [=] {
return self->current_message();
}
};
}
behavior client(event_based_actor* self, actor serv) {
self->send(serv, ok_atom::value);
return {
others >> [=] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
}
};
}
struct server_state {
actor client; // the spawn we connect to the server in our main
actor aut; // our mirror
};
behavior server(stateful_actor<server_state>* self) {
self->on_sync_failure([=] {
CAF_TEST_ERROR("unexpected sync response: "
<< to_string(self->current_message()));
});
return {
[=](ok_atom) {
auto s = self->current_sender();
CAF_REQUIRE(s != invalid_actor_addr);
CAF_REQUIRE(s.is_remote());
self->state.client = actor_cast<actor>(s);
auto mm = io::get_middleman_actor();
self->sync_send(mm, spawn_atom::value,
s.node(), "mirror", make_message()).then(
[=](ok_atom, const actor_addr& addr, const std::set<std::string>& ifs) {
CAF_REQUIRE(addr != invalid_actor_addr);
CAF_CHECK(ifs.empty());
self->state.aut = actor_cast<actor>(addr);
self->send(self->state.aut, "hello mirror");
self->become(
[=](const std::string& str) {
CAF_CHECK(self->current_sender() == self->state.aut);
CAF_CHECK(str == "hello mirror");
self->send_exit(self->state.aut, exit_reason::kill);
self->send_exit(self->state.client, exit_reason::kill);
self->quit();
}
);
},
[=](error_atom, const std::string& errmsg) {
CAF_TEST_ERROR("could not spawn mirror: " << errmsg);
}
);
}
};
}
} // namespace <anonymous>
CAF_TEST(remote_spawn) {
announce_actor_type("mirror", mirror);
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"server,s", "run as server (don't run client"},
{"client,c", "add client port (two needed)", port},
{"port,p", "force a port in server mode", port},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
std::cout << r.error << std::endl << std::endl << r.helptext << std::endl;
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
if (use_asio) {
# ifdef CAF_USE_ASIO
CAF_MESSAGE("enable ASIO backend");
io::set_middleman<io::network::asio_multiplexer>();
# endif // CAF_USE_ASIO
}
if (r.opts.count("client") > 0) {
auto serv = io::remote_actor("localhost", port);
spawn(client, serv);
await_all_actors_done();
return;
}
auto serv = spawn(server);
port = io::publish(serv, port);
CAF_TEST_INFO("published server at port " << port);
if (r.opts.count("server") == 0) {
auto child = detail::run_program(invalid_actor, argv[0], "-n", "-s",
CAF_XSTR(CAF_SUITE), "--", "-c",
port, (use_asio ? "--use-asio" : ""));
child.join();
}
await_all_actors_done();
}
......@@ -219,7 +219,8 @@ CAF_TEST(test_uniform_type) {
"caf::io::connection_closed_msg",
"caf::io::network::protocol",
"caf::io::new_connection_msg",
"caf::io::new_data_msg"));
"caf::io::new_data_msg",
"caf::io::network::address_listing"));
CAF_MESSAGE("io types checked");
}
// check whether enums can be announced as members
......
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