Commit 81f824fb authored by Dominik Charousset's avatar Dominik Charousset

proper networking for typed actors

parent ffed360f
......@@ -10,6 +10,7 @@ __2014_XX_XX__
* New type `exit_msg` is now used instead of messages using the atom `EXIT`
* New type `down_msg` is now used instead of messages using the atom `DOWN`
* New header `system_messages.hpp` for message types used by the runtime
- Announce properly handles empty types
Version 0.8.1
-------------
......
......@@ -34,7 +34,7 @@ cppa/detail/behavior_stack.hpp
cppa/detail/boxed.hpp
cppa/detail/container_tuple_view.hpp
cppa/detail/decorated_tuple.hpp
cppa/detail/default_uniform_type_info_impl.hpp
cppa/detail/default_uniform_type_info.hpp
cppa/detail/demangle.hpp
cppa/detail/disablable_delete.hpp
cppa/detail/empty_tuple.hpp
......@@ -345,6 +345,7 @@ unit_testing/test_serialization.cpp
unit_testing/test_spawn.cpp
unit_testing/test_sync_send.cpp
unit_testing/test_tuple.cpp
unit_testing/test_typed_remote_actor.cpp
unit_testing/test_typed_spawn.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
......@@ -40,7 +40,7 @@
#include "cppa/util/algorithm.hpp"
#include "cppa/util/abstract_uniform_type_info.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
#include "cppa/detail/default_uniform_type_info.hpp"
namespace cppa {
......@@ -117,7 +117,7 @@ const uniform_type_info* announce(const std::type_info& tinfo,
template<class C, class Parent, typename... Ts>
std::pair<C Parent::*, util::abstract_uniform_type_info<C>*>
compound_member(C Parent::*c_ptr, const Ts&... args) {
return {c_ptr, new detail::default_uniform_type_info_impl<C>(args...)};
return {c_ptr, new detail::default_uniform_type_info<C>(args...)};
}
// deals with getter returning a mutable reference
......@@ -132,7 +132,7 @@ compound_member(C Parent::*c_ptr, const Ts&... args) {
template<class C, class Parent, typename... Ts>
std::pair<C& (Parent::*)(), util::abstract_uniform_type_info<C>*>
compound_member(C& (Parent::*getter)(), const Ts&... args) {
return {getter, new detail::default_uniform_type_info_impl<C>(args...)};
return {getter, new detail::default_uniform_type_info<C>(args...)};
}
// deals with getter/setter pair
......@@ -153,31 +153,20 @@ compound_member(const std::pair<GRes (Parent::*)() const,
SRes (Parent::*)(SArg) >& gspair,
const Ts&... args) {
typedef typename util::rm_const_and_ref<GRes>::type mtype;
return {gspair, new detail::default_uniform_type_info_impl<mtype>(args...)};
return {gspair, new detail::default_uniform_type_info<mtype>(args...)};
}
/**
* @brief Adds a new type mapping for @p C to the libcppa type system.
* @tparam C A class that is default constructible, copy constructible,
* and comparable.
* @tparam C A class that is either empty or is default constructible,
* copy constructible, and comparable.
* @param arg First members of @p C.
* @param args Additional members of @p C.
* @warning @p announce is <b>not</b> thead safe!
*/
template<class C, typename... Ts>
inline const uniform_type_info* announce(const Ts&... args) {
auto ptr = new detail::default_uniform_type_info_impl<C>(args...);
return announce(typeid(C), std::unique_ptr<uniform_type_info>{ptr});
}
/**
* @brief Adds a new type mapping for an empty or "tag" class.
* @tparam C A class without any member.
* @warning @p announce_tag is <b>not</b> thead safe!
*/
template<class C>
inline const uniform_type_info* announce_tag() {
auto ptr = new detail::empty_uniform_type_info_impl<C>();
auto ptr = new detail::default_uniform_type_info<C>(args...);
return announce(typeid(C), std::unique_ptr<uniform_type_info>{ptr});
}
......@@ -272,11 +261,11 @@ class meta_cow_tuple : public uniform_type_info {
return (instance) ? any_tuple{*cast(instance)} : any_tuple{};
}
bool equals(const std::type_info& tinfo) const {
bool equal_to(const std::type_info& tinfo) const override {
return typeid(tuple_type) == tinfo;
}
bool equals(const void* instance1, const void* instance2) const {
bool equals(const void* instance1, const void* instance2) const override {
return *cast(instance1) == *cast(instance2);
}
......@@ -303,9 +292,9 @@ class meta_cow_tuple : public uniform_type_info {
/**
* @brief Adds a hint to the type system of libcppa. This type hint can
* significantly increase the network performance, because libcppa
* can use the hint to create tuples with full static type information
* rather than using fully dynamically typed tuples.
* increase the network performance, because libcppa
* can use the hint to create tuples with full static type
* information rather than using fully dynamically typed tuples.
*/
template<typename T, typename... Ts>
inline void announce_tuple() {
......
......@@ -77,6 +77,7 @@
#include "cppa/io/connection_handle.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/actor_registry.hpp"
/**
......@@ -499,6 +500,35 @@ inline void await_all_actors_done() {
get_actor_registry()->await_running_count_equal(0);
}
namespace detail {
void publish_impl(abstract_actor_ptr whom, std::unique_ptr<io::acceptor> aptr);
abstract_actor_ptr remote_actor_impl(io::stream_ptr_pair io,
std::set<std::string> expected_interface);
template<class List>
struct typed_remote_actor_helper;
template<typename... Ts>
struct typed_remote_actor_helper<util::type_list<Ts...>> {
typedef typed_actor<Ts...> return_type;
return_type operator()(io::stream_ptr_pair conn) {
auto iface = return_type::get_interface();
auto tmp = remote_actor_impl(std::move(conn), std::move(iface));
return_type res;
// actually safe, because remote_actor_impl throws on type mismatch
raw_access::unsafe_assign(res, tmp);
return res;
}
return_type operator()(const char* host, std::uint16_t port) {
auto ptr = io::ipv4_io_stream::connect_to(host, port);
return (*this)(io::stream_ptr_pair(ptr, ptr));
}
};
} // namespace detail
/**
* @brief Publishes @p whom at @p port.
*
......@@ -511,6 +541,7 @@ inline void await_all_actors_done() {
*/
void publish(actor whom, std::uint16_t port, const char* addr = nullptr);
// implemented in unicast_network.cpp
/**
* @brief Publishes @p whom using @p acceptor to handle incoming connections.
*
......@@ -520,12 +551,24 @@ void publish(actor whom, std::uint16_t port, const char* addr = nullptr);
*/
void publish(actor whom, std::unique_ptr<io::acceptor> acceptor);
// implemented in unicast_network.cpp
/**
* @brief Establish a new connection to a remote actor via @p connection.
* @param connection A connection to another libcppa process described by a pair
* of input and output stream.
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
* @throws std::invalid_argument Thrown when connecting to a typed actor.
*/
actor remote_actor(io::stream_ptr_pair connection);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
* @param host Valid hostname or IP address.
* @param port TCP port.
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
* @throws std::invalid_argument Thrown when connecting to a typed actor.
*/
actor remote_actor(const char* host, std::uint16_t port);
......@@ -537,13 +580,54 @@ inline actor remote_actor(const std::string& host, std::uint16_t port) {
}
/**
* @brief Establish a new connection to a remote actor via @p connection.
* @param connection A connection to another libcppa process described by a pair
* of input and output stream.
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
* @copydoc publish(actor,std::unique_ptr<io::acceptor>)
*/
actor remote_actor(io::stream_ptr_pair connection);
template<typename... Rs>
void typed_publish(typed_actor<Rs...> whom, std::unique_ptr<io::acceptor> uptr) {
if (!whom) return;
detail::publish_impl(detail::raw_access::get(whom), std::move(uptr));
}
/**
* @copydoc publish(actor,std::uint16_t,const char*)
*/
template<typename... Rs>
void typed_publish(typed_actor<Rs...> whom,
std::uint16_t port, const char* addr = nullptr) {
if (!whom) return;
detail::publish_impl(detail::raw_access::get(whom),
io::ipv4_acceptor::create(port, addr));
}
/**
* @copydoc remote_actor(io::stream_ptr_pair)
*/
template<class List>
typename detail::typed_remote_actor_helper<List>::return_type
typed_remote_actor(io::stream_ptr_pair connection) {
detail::typed_remote_actor_helper<List> f;
return f(std::move(connection));
}
/**
* @copydoc remote_actor(const char*,std::uint16_t)
*/
template<class List>
typename detail::typed_remote_actor_helper<List>::return_type
typed_remote_actor(const char* host, std::uint16_t port) {
detail::typed_remote_actor_helper<List> f;
return f(host, port);
}
/**
* @copydoc remote_actor(const std::string&,std::uint16_t)
*/
template<class List>
typename detail::typed_remote_actor_helper<List>::return_type
typed_remote_actor(const std::string& host, std::uint16_t port) {
detail::typed_remote_actor_helper<List> f;
return f(host.c_str(), port);
}
/**
* @brief Spawns an IO actor of type @p Impl.
......
......@@ -45,11 +45,8 @@ class raw_access {
public:
static abstract_actor* get(const actor& hdl) {
return hdl.m_ptr.get();
}
static abstract_actor* get(const actor_addr& hdl) {
template<typename ActorHandle>
static abstract_actor* get(const ActorHandle& hdl) {
return hdl.m_ptr.get();
}
......@@ -69,6 +66,20 @@ class raw_access {
return {get(hdl)};
}
static actor unsafe_cast(const abstract_actor_ptr& ptr) {
return {ptr.get()};
}
template<typename T>
static void unsafe_assign(T& lhs, const actor& rhs) {
lhs = T{get(rhs)};
}
template<typename T>
static void unsafe_assign(T& lhs, const abstract_actor_ptr& ptr) {
lhs = T{ptr.get()};
}
};
} } // namespace cppa::detail
......
......@@ -40,6 +40,8 @@ class ipv4_io_stream : public stream {
public:
~ipv4_io_stream();
static stream_ptr connect_to(const char* host, std::uint16_t port);
static stream_ptr from_native_socket(native_socket_type fd);
......
......@@ -116,9 +116,11 @@ class middleman {
bool has_reader(continuable* ptr);
/**
* @brief Registers a new peer, i.e., a new node in the network.
* @brief Tries to register a new peer, i.e., a new node in the network.
* Returns false if there is already a connection to @p node,
* otherwise true.
*/
virtual void register_peer(const node_id& node, peer* ptr) = 0;
virtual bool register_peer(const node_id& node, peer* ptr) = 0;
/**
* @brief Returns the peer associated with given node id.
......
......@@ -75,8 +75,8 @@ class peer : public extend<continuable>::with<buffered_writing> {
void enqueue(const message_header& hdr, const any_tuple& msg);
inline bool erase_on_last_proxy_exited() const {
return m_erase_on_last_proxy_exited;
inline bool stop_on_last_proxy_exited() const {
return m_stop_on_last_proxy_exited;
}
inline const node_id& node() const {
......@@ -117,7 +117,7 @@ class peer : public extend<continuable>::with<buffered_writing> {
// if this peer was created using remote_actor(), then m_doorman will
// point to the published actor of the remote node
bool m_erase_on_last_proxy_exited;
bool m_stop_on_last_proxy_exited;
partial_function m_content_handler;
......
......@@ -46,13 +46,16 @@ class peer_acceptor : public continuable {
public:
typedef std::set<std::string> string_set;
continue_reading_result continue_reading() override;
peer_acceptor(middleman* parent,
acceptor_uptr ptr,
const actor_addr& published_actor);
const actor_addr& published_actor,
string_set signatures);
inline const actor_addr& published_actor() const { return m_pa; }
inline const actor_addr& published_actor() const { return m_aa; }
void dispose() override;
......@@ -62,7 +65,8 @@ class peer_acceptor : public continuable {
middleman* m_parent;
acceptor_uptr m_ptr;
actor_addr m_pa;
actor_addr m_aa;
string_set m_sigs;
};
......
......@@ -661,8 +661,9 @@ class match_expr {
static constexpr idx_token_type idx_token = idx_token_type{};
template<typename... Ts>
match_expr(Ts&&... args) : m_cases(std::forward<Ts>(args)...) {
template<typename T, typename... Ts>
match_expr(T arg, Ts&&... args)
: m_cases(std::move(arg), std::forward<Ts>(args)...) {
init();
}
......
......@@ -283,6 +283,14 @@ __unspecified__ on();
template<atom_value... Atoms, typename... Ts>
__unspecified__ on();
/**
* @brief Converts @p arg to a match expression by returning
* <tt>on_arg_match >> arg</tt> if @p arg is a callable type,
* otherwise returns @p arg.
*/
template<typename T>
__unspecified__ lift_to_match_expr(T arg);
#else
template<typename T>
......@@ -408,6 +416,18 @@ class on_the_fly_rvalue_builder {
constexpr detail::on_the_fly_rvalue_builder on_arg_match;
// and even more convenience
template<typename F, class E = typename std::enable_if<util::is_callable<F>::value>::type>
inline auto lift_to_match_expr(F fun) -> decltype(on_arg_match >> fun) {
return (on_arg_match >> fun);
}
template<typename... Cs>
inline match_expr<Cs...> lift_to_match_expr(match_expr<Cs...> expr) {
return std::move(expr);
}
#endif // CPPA_DOCUMENTATION
} // namespace cppa
......
......@@ -47,6 +47,12 @@ struct invalid_actor_addr_t;
template<typename... Rs>
class typed_event_based_actor;
namespace detail {
class raw_access;
} // namespace detail
/**
* @brief Identifies a strongly typed actor.
* @tparam Rs Interface as @p replies_to<...>::with<...> parameter pack.
......@@ -58,6 +64,8 @@ class typed_actor : util::comparable<typed_actor<Rs...>>
friend class local_actor;
friend class detail::raw_access;
// friend with all possible instantiations
template<typename...>
friend class typed_actor;
......@@ -80,7 +88,10 @@ class typed_actor : util::comparable<typed_actor<Rs...>>
*/
typedef typed_event_based_actor<Rs...> base;
typedef util::type_list<Rs...> signatures;
/**
* @brief Stores the interface of the actor as type list.
*/
typedef util::type_list<Rs...> interface;
typed_actor() = default;
typed_actor(typed_actor&&) = default;
......@@ -131,8 +142,22 @@ class typed_actor : util::comparable<typed_actor<Rs...>>
return m_ptr ? 1 : 0;
}
static std::set<std::string> get_interface() {
return {detail::to_uniform_name<Rs>()...};
}
explicit operator bool() const {
return static_cast<bool>(m_ptr);
}
inline bool operator!() const {
return !m_ptr;
}
private:
typed_actor(abstract_actor* ptr) : m_ptr(ptr) { }
template<class ListA, class ListB>
inline void check_signatures() {
static_assert(util::tl_is_strict_subset<ListA, ListB>::value,
......@@ -141,13 +166,13 @@ class typed_actor : util::comparable<typed_actor<Rs...>>
template<typename... OtherRs>
inline void set(const typed_actor<OtherRs...>& other) {
check_signatures<signatures, util::type_list<OtherRs...>>();
check_signatures<interface, util::type_list<OtherRs...>>();
m_ptr = other.m_ptr;
}
template<class Impl>
inline void set(intrusive_ptr<Impl>& other) {
check_signatures<signatures, typename Impl::signatures>();
check_signatures<interface, typename Impl::signatures>();
m_ptr = std::move(other);
}
......
......@@ -133,9 +133,10 @@ class typed_behavior {
typedef util::type_list<Rs...> signatures;
template<typename... Cs, typename... Ts>
typed_behavior(match_expr<Cs...> expr, Ts&&... args) {
set(match_expr_collect(std::move(expr), std::forward<Ts>(args)...));
template<typename T, typename... Ts>
typed_behavior(T arg, Ts&&... args) {
set(match_expr_collect(lift_to_match_expr(std::move(arg)),
lift_to_match_expr(std::forward<Ts>(args))...));
}
template<typename... Cs>
......
......@@ -207,7 +207,7 @@ class uniform_type_info {
* type than @p tinfo.
* @returns @p true if @p tinfo describes the same type as @p this.
*/
virtual bool equals(const std::type_info& tinfo) const = 0;
virtual bool equal_to(const std::type_info& tinfo) const = 0;
/**
* @brief Compares two instances of this type.
......@@ -293,28 +293,28 @@ inline bool operator!=(const uniform_type_info& lhs,
* @relates uniform_type_info
*/
inline bool operator==(const uniform_type_info& lhs, const std::type_info& rhs) {
return lhs.equals(rhs);
return lhs.equal_to(rhs);
}
/**
* @relates uniform_type_info
*/
inline bool operator!=(const uniform_type_info& lhs, const std::type_info& rhs) {
return !(lhs.equals(rhs));
return !(lhs.equal_to(rhs));
}
/**
* @relates uniform_type_info
*/
inline bool operator==(const std::type_info& lhs, const uniform_type_info& rhs) {
return rhs.equals(lhs);
return rhs.equal_to(lhs);
}
/**
* @relates uniform_type_info
*/
inline bool operator!=(const std::type_info& lhs, const uniform_type_info& rhs) {
return !(rhs.equals(lhs));
return !(rhs.equal_to(lhs));
}
} // namespace cppa
......
......@@ -49,7 +49,7 @@ class abstract_uniform_type_info : public uniform_type_info {
public:
bool equals(const std::type_info& tinfo) const {
bool equal_to(const std::type_info& tinfo) const override {
return typeid(T) == tinfo;
}
......@@ -70,20 +70,18 @@ class abstract_uniform_type_info : public uniform_type_info {
else m_name = cname;
}
bool equals(const void* lhs, const void* rhs) const {
return deref(lhs) == deref(rhs);
bool equals(const void* lhs, const void* rhs) const override {
return eq(deref(lhs), deref(rhs));
}
void* new_instance(const void* ptr) const {
void* new_instance(const void* ptr) const override {
return (ptr) ? new T(deref(ptr)) : new T();
}
void delete_instance(void* instance) const {
void delete_instance(void* instance) const override {
delete reinterpret_cast<T*>(instance);
}
private:
static inline const T& deref(const void* ptr) {
return *reinterpret_cast<const T*>(ptr);
}
......@@ -94,6 +92,20 @@ class abstract_uniform_type_info : public uniform_type_info {
std::string m_name;
private:
template<class C>
typename std::enable_if<std::is_empty<C>::value == true , bool>::type
eq(const C&, const C&) const {
return true;
}
template<class C>
typename std::enable_if<std::is_empty<C>::value == false, bool>::type
eq(const C& lhs, const C& rhs) const {
return lhs == rhs;
}
};
} } // namespace cppa::util
......
......@@ -54,6 +54,10 @@ using namespace ::cppa::detail::fd_util;
ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { }
ipv4_io_stream::~ipv4_io_stream() {
closesocket(m_fd);
}
native_socket_type ipv4_io_stream::read_handle() const {
return m_fd;
}
......
......@@ -150,7 +150,7 @@ class middleman_impl : public middleman {
static_cast<void>(res);
}
void register_peer(const node_id& node, peer* ptr) override {
bool register_peer(const node_id& node, peer* ptr) override {
CPPA_LOG_TRACE("node = " << to_string(node) << ", ptr = " << ptr);
auto& entry = m_peers[node];
if (entry.impl == nullptr) {
......@@ -162,10 +162,12 @@ class middleman_impl : public middleman {
ptr->enqueue(tmp.first, tmp.second);
}
CPPA_LOG_INFO("peer " << to_string(node) << " added");
return true;
}
else {
CPPA_LOG_WARNING("peer " << to_string(node) << " already defined, "
"multiple calls to remote_actor()?");
return false;
}
}
......@@ -215,9 +217,8 @@ class middleman_impl : public middleman {
CPPA_REQUIRE(pptr->m_queue != nullptr);
CPPA_LOG_TRACE(CPPA_ARG(pptr)
<< ", pptr->node() = " << to_string(pptr->node()));
if (pptr->erase_on_last_proxy_exited() && pptr->queue().empty()) {
if (pptr->stop_on_last_proxy_exited() && pptr->queue().empty()) {
stop_reader(pptr);
del_peer(pptr);
}
}
......
......@@ -71,7 +71,7 @@ peer::peer(middleman* parent,
: sizeof(uint32_t));
// state == wait_for_msg_size iff peer was created using remote_peer()
// in this case, this peer must be erased if no proxy of it remains
m_erase_on_last_proxy_exited = m_state == wait_for_msg_size;
m_stop_on_last_proxy_exited = m_state == wait_for_msg_size;
m_meta_hdr = uniform_typeid<message_header>();
m_meta_msg = uniform_typeid<any_tuple>();
}
......@@ -119,7 +119,11 @@ continue_reading_result peer::continue_reading() {
return read_failure;
}
CPPA_LOG_DEBUG("read process info: " << to_string(*m_node));
parent()->register_peer(*m_node, this);
if (!parent()->register_peer(*m_node, this)) {
CPPA_LOG_ERROR("multiple incoming connections "
"from the same node");
return read_failure;
}
// initialization done
m_state = wait_for_msg_size;
m_rd_buf.clear();
......@@ -339,7 +343,7 @@ continue_writing_result peer::continue_writing() {
enqueue(tmp.first, tmp.second);
result = super::continue_writing();
}
if (result == write_done && erase_on_last_proxy_exited() && !has_unwritten_data()) {
if (result == write_done && stop_on_last_proxy_exited() && !has_unwritten_data()) {
if (parent()->get_namespace().count_proxies(*m_node) == 0) {
parent()->last_proxy_exited(this);
}
......
......@@ -46,8 +46,10 @@ namespace cppa { namespace io {
peer_acceptor::peer_acceptor(middleman* parent,
acceptor_uptr aur,
const actor_addr& pa)
: super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur)), m_pa(pa) { }
const actor_addr& addr,
string_set sigs)
: super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur))
, m_aa(addr), m_sigs(std::move(sigs)) { }
continue_reading_result peer_acceptor::continue_reading() {
CPPA_LOG_TRACE("");
......@@ -64,11 +66,20 @@ continue_reading_result peer_acceptor::continue_reading() {
auto& pself = node_id::get();
uint32_t process_id = pself->process_id();
try {
auto& out = pair.second;
actor_id aid = published_actor().id();
pair.second->write(&aid, sizeof(actor_id));
pair.second->write(&process_id, sizeof(uint32_t));
pair.second->write(pself->host_id().data(),
// serialize: actor id, process id, node id, interface
out->write(&aid, sizeof(actor_id));
out->write(&process_id, sizeof(uint32_t));
out->write(pself->host_id().data(),
pself->host_id().size());
auto u32_size = static_cast<std::uint32_t>(m_sigs.size());
out->write(&u32_size, sizeof(uint32_t));
for (auto& sig : m_sigs) {
u32_size = static_cast<std::uint32_t>(sig.size());
out->write(&u32_size, sizeof(uint32_t));
out->write(sig.c_str(), sig.size());
}
m_parent->new_peer(pair.first, pair.second);
}
catch (exception& e) {
......
......@@ -73,7 +73,7 @@ remote_actor_proxy::~remote_actor_proxy() {
"node = " << to_string(*node) << ", aid " << aid);
mm->get_namespace().erase(*node, aid);
auto p = mm->get_peer(*node);
if (p && p->erase_on_last_proxy_exited()) {
if (p && p->stop_on_last_proxy_exited()) {
if (mm->get_namespace().count_proxies(*node) == 0) {
mm->last_proxy_exited(p);
}
......
......@@ -60,44 +60,139 @@
#include "cppa/io/ipv4_io_stream.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
namespace {
constexpr std::uint32_t max_iface_size = 100;
constexpr std::uint32_t max_iface_clause_size = 500;
typedef std::set<std::string> string_set;
} // namespace <anonymous>
namespace cppa {
using namespace detail;
using namespace io;
void publish(actor whom, std::unique_ptr<acceptor> aptr) {
void publish(actor whom, std::uint16_t port, const char* addr) {
if (!whom) return;
publish(std::move(whom), io::ipv4_acceptor::create(port, addr));
}
void publish(actor whom, std::unique_ptr<io::acceptor> acceptor) {
detail::publish_impl(detail::raw_access::get(whom), std::move(acceptor));
}
actor remote_actor(io::stream_ptr_pair conn) {
auto res = detail::remote_actor_impl(conn, string_set{});
return detail::raw_access::unsafe_cast(res);
}
actor remote_actor(const char* host, std::uint16_t port) {
auto ptr = ipv4_io_stream::connect_to(host, port);
return remote_actor(stream_ptr_pair(ptr, ptr));
}
namespace detail {
void publish_impl(abstract_actor_ptr ptr, std::unique_ptr<acceptor> aptr) {
// begin the scenes, we serialze/deserialize as actor
actor whom{raw_access::unsafe_cast(ptr.get())};
CPPA_LOGF_TRACE(CPPA_TARG(whom, to_string) << ", " << CPPA_MARG(aptr, get));
if (!whom) return;
get_actor_registry()->put(whom->id(), detail::raw_access::get(whom));
auto mm = get_middleman();
auto addr = whom.address();
mm->register_acceptor(addr, new peer_acceptor(mm, move(aptr), addr));
}
void publish(actor whom, std::uint16_t port, const char* addr) {
if (!whom) return;
publish(whom, ipv4_acceptor::create(port, addr));
auto sigs = whom->interface();
mm->register_acceptor(addr, new peer_acceptor(mm, move(aptr),
addr, move(sigs)));
}
actor remote_actor(stream_ptr_pair io) {
abstract_actor_ptr remote_actor_impl(stream_ptr_pair io, string_set expected) {
CPPA_LOGF_TRACE("io{" << io.first.get() << ", " << io.second.get() << "}");
auto pinf = node_id::get();
std::uint32_t process_id = pinf->process_id();
// throws on error
io.second->write(&process_id, sizeof(std::uint32_t));
io.second->write(pinf->host_id().data(), pinf->host_id().size());
// deserialize: actor id, process id, node id, interface
actor_id remote_aid;
std::uint32_t peer_pid;
node_id::host_id_type peer_node_id;
io.first->read(&remote_aid, sizeof(actor_id));
io.first->read(&peer_pid, sizeof(std::uint32_t));
io.first->read(peer_node_id.data(), peer_node_id.size());
std::uint32_t iface_size;
std::set<std::string> iface;
auto& in = io.first;
// -> actor id
in->read(&remote_aid, sizeof(actor_id));
// -> process id
in->read(&peer_pid, sizeof(std::uint32_t));
// -> node id
in->read(peer_node_id.data(), peer_node_id.size());
// -> interface
in->read(&iface_size, sizeof(std::uint32_t));
if (iface_size > max_iface_size) {
throw std::invalid_argument("Remote actor claims to have more than"
+std::to_string(max_iface_size)+
" message types? Someone is trying"
" something nasty!");
}
std::vector<char> strbuf;
for (std::uint32_t i = 0; i < iface_size; ++i) {
std::uint32_t str_size;
in->read(&str_size, sizeof(std::uint32_t));
if (str_size > max_iface_clause_size) {
throw std::invalid_argument("Remote actor claims to have a"
" reply_to<...>::with<...> clause with"
" more than"
+std::to_string(max_iface_clause_size)+
" characters? Someone is"
" trying something nasty!");
}
strbuf.reserve(str_size + 1);
strbuf.resize(str_size);
in->read(strbuf.data(), str_size);
strbuf.push_back('\0');
iface.insert(std::string{strbuf.data()});
}
// deserialization done, check interface
if (iface != expected) {
auto tostr = [](const std::set<std::string>& what) -> std::string {
if (what.empty()) return "actor";
std::string tmp;
tmp = "typed_actor<";
auto i = what.begin();
auto e = what.end();
tmp += *i++;
while (i != e) tmp += *i++;
tmp += ">";
return tmp;
};
auto iface_str = tostr(iface);
auto expected_str = tostr(expected);
if (expected.empty()) {
throw std::invalid_argument("expected remote actor to be a "
"dynamically typed actor but found "
"a strongly typed actor of type "
+ iface_str);
}
if (iface.empty()) {
throw std::invalid_argument("expected remote actor to be a "
"strongly typed actor of type "
+ expected_str +
" but found a dynamically typed actor");
}
throw std::invalid_argument("expected remote actor to be a "
"strongly typed actor of type "
+ expected_str +
" but found a strongly typed actor of type "
+ iface_str);
}
auto pinfptr = make_counted<node_id>(peer_pid, peer_node_id);
if (*pinf == *pinfptr) {
// this is a local actor, not a remote actor
CPPA_LOGF_WARNING("remote_actor() called to access a local actor");
auto ptr = get_actor_registry()->get(remote_aid);
return detail::raw_access::unsafe_cast(ptr.get());
return ptr;
}
auto mm = get_middleman();
struct remote_actor_result { remote_actor_result* next; actor value; };
......@@ -113,12 +208,8 @@ actor remote_actor(stream_ptr_pair io) {
});
std::unique_ptr<remote_actor_result> result(q.pop());
CPPA_LOGF_DEBUG(CPPA_MARG(result, get));
return result->value;
}
actor remote_actor(const char* host, std::uint16_t port) {
auto io = ipv4_io_stream::connect_to(host, port);
return remote_actor(stream_ptr_pair(io, io));
return raw_access::get(result->value);
}
} // namespace detail
} // namespace cppa
......@@ -57,7 +57,6 @@
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
namespace cppa {
......
......@@ -53,7 +53,7 @@
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
#include "cppa/detail/default_uniform_type_info.hpp"
namespace cppa { namespace detail {
......@@ -438,19 +438,19 @@ class uti_base : public uniform_type_info {
uti_base() : m_native(&typeid(T)) { }
bool equals(const std::type_info& ti) const {
bool equal_to(const std::type_info& ti) const override {
return types_equal(m_native, &ti);
}
bool equals(const void* lhs, const void* rhs) const {
bool equals(const void* lhs, const void* rhs) const override {
return deref(lhs) == deref(rhs);
}
void* new_instance(const void* ptr) const {
void* new_instance(const void* ptr) const override {
return (ptr) ? new T(deref(ptr)) : new T;
}
void delete_instance(void* instance) const {
void delete_instance(void* instance) const override {
delete reinterpret_cast<T*>(instance);
}
......@@ -516,15 +516,15 @@ class int_tinfo : public abstract_int_tinfo {
public:
void serialize(const void* instance, serializer* sink) const {
void serialize(const void* instance, serializer* sink) const override {
sink->write_value(deref(instance));
}
void deserialize(void* instance, deserializer* source) const {
void deserialize(void* instance, deserializer* source) const override {
deref(instance) = source->read<T>();
}
const char* name() const {
const char* name() const override {
return static_name();
}
......@@ -534,7 +534,7 @@ class int_tinfo : public abstract_int_tinfo {
protected:
bool equals(const std::type_info& ti) const {
bool equal_to(const std::type_info& ti) const override {
auto tptr = &ti;
return std::any_of(m_natives.begin(), m_natives.end(),
[tptr](const std::type_info* ptr) {
......@@ -542,15 +542,15 @@ class int_tinfo : public abstract_int_tinfo {
});
}
bool equals(const void* lhs, const void* rhs) const {
bool equals(const void* lhs, const void* rhs) const override {
return deref(lhs) == deref(rhs);
}
void* new_instance(const void* ptr) const {
void* new_instance(const void* ptr) const override {
return (ptr) ? new T(deref(ptr)) : new T;
}
void delete_instance(void* instance) const {
void delete_instance(void* instance) const override {
delete reinterpret_cast<T*>(instance);
}
......@@ -595,11 +595,11 @@ class buffer_type_info_impl : public uniform_type_info {
protected:
bool equals(const std::type_info& ti) const {
bool equal_to(const std::type_info& ti) const override {
return ti == typeid(util::buffer);
}
bool equals(const void* vlhs, const void* vrhs) const {
bool equals(const void* vlhs, const void* vrhs) const override {
auto& lhs = deref(vlhs);
auto& rhs = deref(vrhs);
return (lhs.empty() && rhs.empty())
......@@ -607,11 +607,11 @@ class buffer_type_info_impl : public uniform_type_info {
&& memcmp(lhs.data(), rhs.data(), lhs.size()) == 0);
}
void* new_instance(const void* ptr) const {
void* new_instance(const void* ptr) const override {
return (ptr) ? new util::buffer(deref(ptr)) : new util::buffer;
}
void delete_instance(void* instance) const {
void delete_instance(void* instance) const override {
delete reinterpret_cast<util::buffer*>(instance);
}
......@@ -686,7 +686,7 @@ class default_meta_tuple : public uniform_type_info {
}
}
bool equals(const std::type_info&) const override {
bool equal_to(const std::type_info&) const override {
return false;
}
......@@ -697,7 +697,6 @@ class default_meta_tuple : public uniform_type_info {
return std::equal(lhs.begin(), lhs.end(), rhs.begin(), cmp);
}
private:
std::string m_name;
......@@ -920,7 +919,7 @@ class utim_impl : public uniform_type_info_map {
// 20-29
uti_impl<std::u32string> m_type_u32str;
default_uniform_type_info_impl<strmap> m_type_strmap;
default_uniform_type_info<strmap> m_type_strmap;
uti_impl<bool> m_type_bool;
uti_impl<float> m_type_float;
uti_impl<double> m_type_double;
......@@ -945,7 +944,7 @@ class utim_impl : public uniform_type_info_map {
pointer find_rtti(const Container& c, const std::type_info& ti) const {
auto e = c.end();
auto i = std::find_if(c.begin(), e, [&](pointer p) {
return p->equals(ti);
return p->equal_to(ti);
});
return (i == e) ? nullptr : *i;
}
......
......@@ -28,6 +28,7 @@ add_unit_test(typed_spawn)
add_unit_test(local_group)
add_unit_test(sync_send)
add_unit_test(remote_actor ping_pong.cpp)
add_unit_test(typed_remote_actor)
add_unit_test(broker)
if (ENABLE_OPENCL)
......
......@@ -383,28 +383,28 @@ int main(int argc, char** argv) {
do {
CPPA_TEST(test_remote_actor);
thread child;
ostringstream oss;
if (run_remote_actor) {
oss << app_path << " run=remote_actor port=" << port << " &>/dev/null";
// execute client_part() in a separate process,
// connected via localhost socket
child = thread([&oss]() {
CPPA_LOGC_TRACE("NONE", "main$thread_launcher", "");
string cmdstr = oss.str();
if (system(cmdstr.c_str()) != 0) {
CPPA_PRINTERR("FATAL: command \"" << cmdstr << "\" failed!");
abort();
}
});
}
else { CPPA_PRINT("actor published at port " << port); }
CPPA_CHECKPOINT();
self->receive (
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.source, serv);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
ostringstream oss;
if (run_remote_actor) {
oss << app_path << " run=remote_actor port=" << port << " &>/dev/null";
// execute client_part() in a separate process,
// connected via localhost socket
child = thread([&oss]() {
CPPA_LOGC_TRACE("NONE", "main$thread_launcher", "");
string cmdstr = oss.str();
if (system(cmdstr.c_str()) != 0) {
CPPA_PRINTERR("FATAL: command \"" << cmdstr << "\" failed!");
abort();
}
);
});
}
else { CPPA_PRINT("actor published at port " << port); }
CPPA_CHECKPOINT();
self->receive (
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.source, serv);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
// wait until separate process (in sep. thread) finished execution
CPPA_CHECKPOINT();
if (run_remote_actor) child.join();
......
......@@ -53,7 +53,6 @@
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/type_to_ptype.hpp"
#include "cppa/detail/ptype_to_type.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
using namespace std;
using namespace cppa;
......@@ -123,18 +122,21 @@ bool operator!=(const raw_struct& lhs, const raw_struct& rhs) {
}
struct raw_struct_type_info : util::abstract_uniform_type_info<raw_struct> {
void serialize(const void* ptr, serializer* sink) const {
void serialize(const void* ptr, serializer* sink) const override {
auto rs = reinterpret_cast<const raw_struct*>(ptr);
sink->write_value(static_cast<uint32_t>(rs->str.size()));
sink->write_raw(rs->str.size(), rs->str.data());
}
void deserialize(void* ptr, deserializer* source) const {
void deserialize(void* ptr, deserializer* source) const override {
auto rs = reinterpret_cast<raw_struct*>(ptr);
rs->str.clear();
auto size = source->read<std::uint32_t>();
rs->str.resize(size);
source->read_raw(size, &(rs->str[0]));
}
bool equals(const void* lhs, const void* rhs) const override {
return deref(lhs) == deref(rhs);
}
};
void test_ieee_754() {
......
#include <thread>
#include <string>
#include <cstring>
#include <sstream>
#include <iostream>
#include <functional>
#include "test.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
struct ping { std::int32_t value; };
bool operator==(const ping& lhs, const ping& rhs) {
return lhs.value == rhs.value;
}
struct pong { std::int32_t value; };
bool operator==(const pong& lhs, const pong& rhs) {
return lhs.value == rhs.value;
}
using server_type = typed_actor<
replies_to<ping>::with<pong>
>;
typedef typed_actor<> client_type;
server_type::behavior_type server() {
return {
[](const ping& p) -> pong {
CPPA_CHECKPOINT();
return pong{p.value};
}
};
}
void run_client(const char* host, std::uint16_t port) {
// check whether invalid_argument is thrown
// when trying to connect to get an untyped
// handle to the server
try {
remote_actor(host, port);
}
catch (std::invalid_argument& e) {
cout << e.what() << endl;
CPPA_CHECKPOINT();
}
catch (std::exception& e) {
cerr << "unexpected: " << e.what() << endl;
}
CPPA_CHECKPOINT();
auto serv = typed_remote_actor<server_type::interface>(host, port);
CPPA_CHECKPOINT();
scoped_actor self;
self->sync_send(serv, ping{42}).await(
[](const pong& p) {
CPPA_CHECK_EQUAL(p.value, 42);
}
);
anon_send_exit(serv, exit_reason::user_shutdown);
}
std::uint16_t run_server() {
auto ref = spawn_typed(server);
uint16_t port = 4242;
for (;;) {
try {
typed_publish(ref, port, "127.0.0.1");
CPPA_LOGF_DEBUG("running on port " << port);
return port;
}
catch (bind_failure&) {
// try next port
++port;
}
}
}
int main(int argc, char** argv) {
announce<ping>(&ping::value);
announce<pong>(&pong::value);
auto run_remote_actor = true;
if (argc > 1) {
if (strcmp(argv[1], "run_remote_actor=false") == 0) {
CPPA_LOGF_INFO("don't run remote actor");
run_remote_actor = false;
}
else {
auto kvp = get_kv_pairs(argc, argv);
if (kvp.count("port") == 0) {
throw std::invalid_argument("no port given");
}
run_client("localhost",
static_cast<std::uint16_t>(std::stoi(kvp["port"])));
CPPA_CHECKPOINT();
await_all_actors_done();
shutdown();
return CPPA_TEST_RESULT();
}
}
CPPA_CHECKPOINT();
auto port = run_server();
CPPA_CHECKPOINT();
if (run_remote_actor) {
CPPA_CHECKPOINT();
thread child;
ostringstream oss;
oss << argv[0] << " run=remote_actor port=" << port << " &>/dev/null";
// execute client_part() in a separate process,
// connected via localhost socket
child = thread([&oss]() {
CPPA_LOGC_TRACE("NONE", "main$thread_launcher", "");
string cmdstr = oss.str();
if (system(cmdstr.c_str()) != 0) {
CPPA_PRINTERR("FATAL: command \"" << cmdstr << "\" failed!");
abort();
}
});
CPPA_CHECKPOINT();
child.join();
}
else {
CPPA_PRINT("actor published at port " << port);
}
CPPA_CHECKPOINT();
await_all_actors_done();
shutdown();
return CPPA_TEST_RESULT();
}
......@@ -316,7 +316,7 @@ int main() {
CPPA_TEST(test_typed_spawn);
// announce stuff
announce_tag<get_state_msg>();
announce<get_state_msg>();
announce<int_actor>();
announce<my_request>(&my_request::a, &my_request::b);
......
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