Commit f493412e authored by Dominik Charousset's avatar Dominik Charousset

streamlined middleman layer

this patch removes the superfluous `protocol` class, replaces
`actor_addressing`/`default_actor_addressing` with `actor_namespace`
and removes the `default` suffix from IO classes
parent 4da53f18
......@@ -104,7 +104,7 @@ set(LIBCPPA_SRC
${LIBCPPA_PLATFORM_SRC}
src/abstract_tuple.cpp
src/actor.cpp
src/actor_addressing.cpp
src/actor_namespace.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/algorithm.cpp
......@@ -121,11 +121,9 @@ set(LIBCPPA_SRC
src/context_switching_actor.cpp
src/continuable.cpp
src/decorated_tuple.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
src/default_peer.cpp
src/default_peer_acceptor.cpp
src/default_protocol.cpp
src/actor_proxy.cpp
src/peer.cpp
src/peer_acceptor.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
......@@ -158,8 +156,8 @@ set(LIBCPPA_SRC
src/partial_function.cpp
src/primitive_variant.cpp
src/process_information.cpp
src/protocol.cpp
src/ref_counted.cpp
src/remote_actor_proxy.cpp
src/response_handle.cpp
src/ripemd_160.cpp
src/scheduled_actor.cpp
......@@ -247,7 +245,7 @@ endif ()
# bulid shared library only if not comiling static only
if (NOT "${CPPA_BUILD_STATIC_ONLY}" STREQUAL "yes")
add_library(libcppa SHARED ${LIBCPPA_HDRS} ${LIBCPPA_SRC})
add_library(libcppa SHARED ${LIBCPPA_SRC} ${LIBCPPA_HDRS})
target_link_libraries(libcppa ${LD_FLAGS})
set(LIBRARY_VERSION ${LIBCPPA_VERSION_MAJOR}.${LIBCPPA_VERSION_MINOR}.${LIBCPPA_VERSION_PATCH})
set(LIBRARY_SOVERSION ${LIBCPPA_VERSION_MAJOR})
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_ACTOR_PROXY_CACHE_HPP
#define CPPA_ACTOR_PROXY_CACHE_HPP
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/ref_counted.hpp"
namespace cppa {
class serializer;
class deserializer;
/**
* @brief Different serialization protocols have different representations
* for actors. This class encapsulates a technology-specific
* actor addressing.
*/
class actor_addressing {
public:
virtual ~actor_addressing();
/**
* @brief Returns the technology identifier of the implementation.
* @note All-uppercase identifiers are reserved for libcppa's
* builtin implementations.
*/
virtual atom_value technology_id() const = 0;
/**
* @brief Serializes @p ptr to @p sink according
* to the implemented addressing.
* @warning Not thread-safe
*/
virtual void write(serializer* sink, const actor_ptr& ptr) = 0;
/**
* @brief Deserializes an actor from @p source according
* to the implemented addressing.
* @warning Not thread-safe
*/
virtual actor_ptr read(deserializer* source) = 0;
};
} // namespace cppa::detail
#endif // CPPA_ACTOR_PROXY_CACHE_HPP
......@@ -25,62 +25,112 @@
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
\******************************************************************************/
#ifndef CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
#define CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
#ifndef CPPA_ACTOR_NAMESPACE_HPP
#define CPPA_ACTOR_NAMESPACE_HPP
#include <map>
#include <cstdint>
#include <utility>
#include <functional>
#include "cppa/actor_proxy.hpp"
#include "cppa/actor_addressing.hpp"
#include "cppa/process_information.hpp"
namespace cppa { namespace io {
namespace cppa {
class default_protocol;
class serializer;
class deserializer;
class default_actor_addressing : public actor_addressing {
namespace io { class middleman; }
public:
default_actor_addressing(default_protocol* parent = nullptr);
/**
* @brief Groups a (distributed) set of actors and allows actors
* in the same namespace to exchange messages.
*/
class actor_namespace {
typedef std::map<actor_id, weak_actor_proxy_ptr> proxy_map;
public:
typedef std::function<actor_proxy_ptr(actor_id, process_information_ptr)>
factory_fun;
atom_value technology_id() const;
typedef std::function<void(actor_id, const process_information&)>
new_element_callback;
inline void set_proxy_factory(factory_fun fun);
inline void set_new_element_callback(new_element_callback fun);
void write(serializer* sink, const actor_ptr& ptr);
actor_ptr read(deserializer* source);
// returns the number of proxy instances for given parent
size_t count_proxies(const process_information& parent);
actor_ptr get(const process_information& parent, actor_id aid);
actor_ptr get_or_put(const process_information& parent, actor_id aid);
/**
* @brief A map that stores weak actor proxy pointers by actor ids.
*/
typedef std::map<actor_id, weak_actor_proxy_ptr> proxy_map;
/**
* @brief Returns the number of proxies for @p node.
*/
size_t count_proxies(const process_information& node);
/**
* @brief Returns the proxy instance identified by @p node and @p aid
* or @p nullptr if the actor is unknown.
*/
actor_ptr get(const process_information& node, actor_id aid);
/**
* @brief Returns the proxy instance identified by @p node and @p aid
* or creates a new (default) proxy instance.
*/
actor_ptr get_or_put(process_information_ptr node, actor_id aid);
/**
* @brief Stores @p proxy in the list of known actor proxies.
*/
void put(const process_information& parent,
actor_id aid,
const actor_proxy_ptr& proxy);
proxy_map& proxies(process_information& from);
void erase(process_information& info);
void erase(process_information& info, actor_id aid);
/**
* @brief Returns the map of known actors for @p node.
*/
proxy_map& proxies(process_information& node);
/**
* @brief Deletes all proxies for @p node.
*/
void erase(process_information& node);
/**
* @brief Deletes the proxy with id @p aid for @p node.
*/
void erase(process_information& node, actor_id aid);
private:
default_protocol* m_parent;
process_information_ptr m_pinf;
factory_fun m_factory;
new_element_callback m_new_element_callback;
process_information_ptr m_node;
std::map<process_information, proxy_map> m_proxies;
};
} } // namespace cppa::network
#endif // CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
inline void actor_namespace::set_proxy_factory(factory_fun fun) {
m_factory = std::move(fun);
}
inline void actor_namespace::set_new_element_callback(new_element_callback fun) {
m_new_element_callback = std::move(fun);
}
} // namespace cppa
#endif
......@@ -46,11 +46,11 @@ class binary_deserializer : public deserializer {
public:
binary_deserializer(const void* buf, size_t buf_size,
actor_addressing* addressing = nullptr,
actor_namespace* ns = nullptr,
type_lookup_table* table = nullptr);
binary_deserializer(const void* begin, const void* m_end,
actor_addressing* addressing = nullptr,
actor_namespace* ns = nullptr,
type_lookup_table* table = nullptr);
const uniform_type_info* begin_object() override;
......
......@@ -55,7 +55,7 @@ class binary_serializer : public serializer {
* @warning @p write_buffer must be guaranteed to outlive @p this
*/
binary_serializer(util::buffer* write_buffer,
actor_addressing* addressing = nullptr,
actor_namespace* ns = nullptr,
type_lookup_table* lookup_table = nullptr);
void begin_object(const uniform_type_info*) override;
......
......@@ -42,6 +42,7 @@ class channel;
class behavior;
class any_tuple;
class self_type;
class actor_proxy;
class message_header;
class partial_function;
class uniform_type_info;
......@@ -58,13 +59,18 @@ enum class atom_value : std::uint64_t;
// class templates
template<typename> class optional;
template<typename> class intrusive_ptr;
template<typename> class weak_intrusive_ptr;
// typedefs
// intrusive pointer typedefs
typedef intrusive_ptr<actor> actor_ptr;
typedef intrusive_ptr<group> group_ptr;
typedef intrusive_ptr<channel> channel_ptr;
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
typedef intrusive_ptr<process_information> process_information_ptr;
// weak intrusive pointer typedefs
typedef weak_intrusive_ptr<actor_proxy> weak_actor_proxy_ptr;
} // namespace cppa
#endif // CPPA_FWD_HPP
......@@ -40,7 +40,7 @@
namespace cppa {
class object;
class actor_addressing;
class actor_namespace;
class type_lookup_table;
namespace util { class buffer; }
......@@ -56,7 +56,7 @@ class deserializer {
public:
deserializer(actor_addressing* addressing = nullptr,
deserializer(actor_namespace* ns = nullptr,
type_lookup_table* incoming_types = nullptr);
virtual ~deserializer();
......@@ -123,8 +123,8 @@ class deserializer {
*/
virtual void read_raw(size_t num_bytes, void* storage) = 0;
inline actor_addressing* addressing() {
return m_addressing;
inline actor_namespace* get_namespace() {
return m_namespace;
}
inline type_lookup_table* incoming_types() {
......@@ -135,7 +135,7 @@ class deserializer {
private:
actor_addressing* m_addressing;
actor_namespace* m_namespace;
type_lookup_table* m_incoming_types;
};
......
......@@ -50,7 +50,7 @@ class buffered_writing : public Base {
template<typename... Ts>
buffered_writing(middleman* mm, output_stream_ptr out, Ts&&... args)
: super{std::forward<Ts>(args)...}, m_middleman{mm}, m_out{out}
: super{std::forward<Ts>(args)...}, m_parent{mm}, m_out{out}
, m_has_unwritten_data{false} { }
continue_writing_result continue_writing() override {
......@@ -105,7 +105,7 @@ class buffered_writing : public Base {
if (!m_has_unwritten_data) {
CPPA_LOG_DEBUG("register for writing");
m_has_unwritten_data = true;
m_middleman->continue_writer(this);
m_parent->continue_writer(this);
}
}
......@@ -115,11 +115,15 @@ class buffered_writing : public Base {
protected:
inline middleman* parent() {
return m_parent;
}
typedef buffered_writing combined_type;
private:
middleman* m_middleman;
middleman* m_parent;
output_stream_ptr m_out;
bool m_has_unwritten_data;
util::buffer m_buf;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef DEFAULT_PROTOCOL_HPP
#define DEFAULT_PROTOCOL_HPP
#include <map>
#include <vector>
#include "cppa/actor_addressing.hpp"
#include "cppa/process_information.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/default_peer_acceptor.hpp"
#include "cppa/io/default_message_queue.hpp"
#include "cppa/io/default_actor_addressing.hpp"
namespace cppa { namespace io {
class default_protocol : public protocol {
typedef protocol super;
friend class default_peer;
friend class default_peer_acceptor;
public:
default_protocol(middleman* multiplexer);
atom_value identifier() const;
void publish(const actor_ptr& whom, variant_args args);
void publish(const actor_ptr& whom,
std::unique_ptr<acceptor> acceptor,
variant_args args );
void unpublish(const actor_ptr& whom);
actor_ptr remote_actor(variant_args args);
actor_ptr remote_actor(stream_ptr_pair ioptrs, variant_args args);
void register_peer(const process_information& node, default_peer* ptr);
default_peer* get_peer(const process_information& node);
void enqueue(const process_information& node,
const message_header& hdr,
any_tuple msg);
void new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node = nullptr);
void last_proxy_exited(default_peer* pptr);
void continue_writer(default_peer* pptr);
// covariant return type
default_actor_addressing* addressing();
private:
void del_acceptor(default_peer_acceptor* ptr);
void del_peer(default_peer* ptr);
struct peer_entry {
default_peer* impl;
default_message_queue_ptr queue;
};
default_actor_addressing m_addressing;
std::map<actor_ptr, std::vector<default_peer_acceptor*>> m_acceptors;
std::map<process_information, peer_entry> m_peers;
};
} } // namespace cppa::network
#endif // DEFAULT_PROTOCOL_HPP
......@@ -36,17 +36,36 @@
#include <memory>
#include <functional>
#include "cppa/io/continuable.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/process_information.hpp"
namespace cppa { namespace detail { class singleton_manager; } }
namespace cppa {
class actor_proxy;
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
typedef weak_intrusive_ptr<actor_proxy> weak_actor_proxy_ptr;
} // namespace cppa
namespace cppa { namespace io {
class protocol;
class middleman_impl;
class peer;
class continuable;
class input_stream;
class peer_acceptor;
class output_stream;
class middleman_event_handler;
typedef intrusive_ptr<input_stream> input_stream_ptr;
typedef intrusive_ptr<output_stream> output_stream_ptr;
/**
* @brief Multiplexes asynchronous IO.
* @note No member function except for @p run_later is safe to call from
* outside the event loop.
*/
class middleman {
......@@ -55,26 +74,20 @@ class middleman {
public:
virtual ~middleman();
/**
* @brief Returns the networking protocol in use.
* @brief Runs @p fun in the event loop of the middleman.
* @note This member function is thread-safe.
*/
protocol* get_protocol();
/**
* @brief Runs @p fun in the middleman's event loop.
*/
void run_later(std::function<void()> fun);
virtual void run_later(std::function<void()> fun) = 0;
/**
* @brief Removes @p ptr from the list of active writers.
* @warning This member function is not thread-safe.
*/
void stop_writer(continuable* ptr);
/**
* @brief Adds @p ptr to the list of active writers.
* @warning This member function is not thread-safe.
*/
void continue_writer(continuable* ptr);
......@@ -102,30 +115,84 @@ class middleman {
*/
bool has_reader(continuable* ptr);
protected:
// destroys singleton
void destroy();
// initializes singletons
void initialize();
private:
// sets m_impl and binds implementation to given protocol
void set_pimpl(std::unique_ptr<protocol>&&);
/**
* @brief Registers a new peer, i.e., a new node in the network.
*/
virtual void register_peer(const process_information& node, peer* ptr) = 0;
/**
* @brief Returns the peer associated with given node id.
*/
virtual peer* get_peer(const process_information& node) = 0;
/**
* @brief This callback is used by peer_acceptor implementations to
* invoke cleanup code when disposed.
*/
virtual void del_acceptor(peer_acceptor* ptr) = 0;
/**
* @brief Delivers a message to given node.
*/
virtual void deliver(const process_information& node,
const message_header& hdr,
any_tuple msg ) = 0;
/**
* @brief This callback is invoked by {@link peer} implementations
* and causes the middleman to disconnect from the node.
*/
virtual void last_proxy_exited(peer* ptr) = 0;
/**
*
*/
virtual void new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node = nullptr) = 0;
// creates a middleman using io::default_protocol
/**
* @brief Adds a new acceptor for incoming connections to @p pa
* to the event loop of the middleman.
* @note This member function is thread-safe.
*/
virtual void register_acceptor(const actor_ptr& pa, peer_acceptor* ptr) = 0;
/**
* @brief Returns the namespace that contains all remote actors
* connected to this middleman.
*/
inline actor_namespace& get_namespace();
protected:
// creates a middleman instance
static middleman* create_singleton();
// destroys uninitialized instances
inline void dispose() { delete this; }
// pointer to implementation
std::unique_ptr<middleman_impl> m_impl;
// destroys an initialized singleton
virtual void destroy() = 0;
// initializes a singleton
virtual void initialize() = 0;
// each middleman defines its own namespace
actor_namespace m_namespace;
// the node id of this middleman
process_information_ptr m_node;
//
std::unique_ptr<middleman_event_handler> m_handler;
};
} } // namespace cppa::detail
inline actor_namespace& middleman::get_namespace() {
return m_namespace;
}
} } // namespace cppa::io
#endif // MIDDLEMAN_HPP
......@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef CPPA_DEFAULT_PEER_IMPL_HPP
#define CPPA_DEFAULT_PEER_IMPL_HPP
#ifndef CPPA_peer_IMPL_HPP
#define CPPA_peer_IMPL_HPP
#include <map>
#include <cstdint>
......@@ -50,20 +50,20 @@
namespace cppa { namespace io {
class default_protocol;
class default_peer : public extend<continuable>::with<buffered_writing> {
class middleman_impl;
class peer : public extend<continuable>::with<buffered_writing> {
typedef combined_type super;
friend class default_protocol;
friend class middleman_impl;
public:
default_peer(default_protocol* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr = nullptr);
peer(middleman* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr = nullptr);
continue_reading_result continue_reading() override;
......@@ -94,7 +94,6 @@ class default_peer : public extend<continuable>::with<buffered_writing> {
read_message
};
default_protocol* m_parent;
input_stream_ptr m_in;
read_state m_state;
process_information_ptr m_node;
......@@ -146,4 +145,4 @@ class default_peer : public extend<continuable>::with<buffered_writing> {
} } // namespace cppa::network
#endif // CPPA_DEFAULT_PEER_IMPL_HPP
#endif // CPPA_peer_IMPL_HPP
......@@ -40,7 +40,7 @@ namespace cppa { namespace io {
class default_protocol;
class default_peer_acceptor : public continuable {
class peer_acceptor : public continuable {
typedef continuable super;
......@@ -48,7 +48,7 @@ class default_peer_acceptor : public continuable {
continue_reading_result continue_reading() override;
default_peer_acceptor(default_protocol* parent,
peer_acceptor(middleman* parent,
acceptor_uptr ptr,
const actor_ptr& published_actor);
......@@ -60,9 +60,9 @@ class default_peer_acceptor : public continuable {
private:
default_protocol* m_parent;
middleman* m_parent;
acceptor_uptr m_ptr;
actor_ptr m_pa;
actor_ptr m_pa;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_PROTOCOL_HPP
#define CPPA_PROTOCOL_HPP
#include <memory>
#include <functional>
#include <initializer_list>
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp"
namespace cppa { class actor_addressing; }
namespace cppa { namespace io {
class continuable;
class abstract_middleman;
/**
* @brief Implements a communication protocol.
*/
class protocol {
public:
typedef std::initializer_list<primitive_variant> variant_args;
protocol(middleman* parent);
virtual atom_value identifier() const = 0;
virtual void publish(const actor_ptr& whom, variant_args args) = 0;
virtual void publish(const actor_ptr& whom,
std::unique_ptr<acceptor> acceptor,
variant_args args ) = 0;
virtual void unpublish(const actor_ptr& whom) = 0;
virtual actor_ptr remote_actor(variant_args args) = 0;
virtual actor_ptr remote_actor(stream_ptr_pair ioptrs,
variant_args args ) = 0;
virtual actor_addressing* addressing() = 0;
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
template<typename T>
inline void run_later(T&& what);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_writer(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_writer(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_reader(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_reader(continuable* ptr);
/**
* @brief Returns the parent of this protocol instance.
*/
inline middleman* parent();
private:
middleman* m_parent;
};
typedef intrusive_ptr<protocol> protocol_ptr;
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline middleman* protocol::parent() {
return m_parent;
}
template<typename T>
inline void protocol::run_later(T&& what) {
m_parent->run_later(std::forward<T>(what));
}
inline void protocol::stop_writer(continuable* ptr) {
m_parent->stop_writer(ptr);
}
inline void protocol::continue_writer(continuable* ptr) {
m_parent->continue_writer(ptr);
}
inline void protocol::stop_reader(continuable* ptr) {
m_parent->stop_reader(ptr);
}
inline void protocol::continue_reader(continuable* ptr) {
m_parent->continue_reader(ptr);
}
} } // namespace cppa::network
#endif // CPPA_PROTOCOL_HPP
......@@ -28,13 +28,12 @@
\******************************************************************************/
#ifndef DEFAULT_ACTOR_PROXY_HPP
#define DEFAULT_ACTOR_PROXY_HPP
#ifndef REMOTE_ACTOR_PROXY_HPP
#define REMOTE_ACTOR_PROXY_HPP
#include "cppa/extend.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
......@@ -48,6 +47,8 @@ class basic_memory_cache;
namespace cppa { namespace io {
class middleman;
class sync_request_info : public extend<memory_managed>::with<memory_cached> {
friend class detail::memory;
......@@ -66,15 +67,15 @@ class sync_request_info : public extend<memory_managed>::with<memory_cached> {
};
class default_actor_proxy : public actor_proxy {
class remote_actor_proxy : public actor_proxy {
typedef actor_proxy super;
public:
default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
default_protocol* parent);
remote_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
middleman* parent);
void enqueue(const message_header& hdr, any_tuple msg) override;
......@@ -98,13 +99,13 @@ class default_actor_proxy : public actor_proxy {
protected:
~default_actor_proxy();
~remote_actor_proxy();
private:
void forward_msg(const message_header& hdr, any_tuple msg);
default_protocol* m_parent;
middleman* m_parent;
process_information_ptr m_pinf;
intrusive::single_reader_queue<sync_request_info, detail::disposer> m_pending_requests;
......@@ -112,4 +113,4 @@ class default_actor_proxy : public actor_proxy {
} } // namespace cppa::network
#endif // DEFAULT_ACTOR_PROXY_HPP
#endif // remote_actor_proxy_HPP
......@@ -39,7 +39,7 @@
namespace cppa {
class actor_addressing;
class actor_namespace;
class primitive_variant;
class type_lookup_table;
......@@ -57,7 +57,7 @@ class serializer {
/**
* @note @p addressing must be guaranteed to outlive the serializer
*/
serializer(actor_addressing* addressing = nullptr,
serializer(actor_namespace* addressing = nullptr,
type_lookup_table* outgoing_types = nullptr);
virtual ~serializer();
......@@ -104,8 +104,8 @@ class serializer {
*/
virtual void write_tuple(size_t num, const primitive_variant* values) = 0;
inline actor_addressing* addressing() {
return m_addressing;
inline actor_namespace* get_namespace() {
return m_namespace;
}
inline type_lookup_table* outgoing_types() {
......@@ -114,7 +114,7 @@ class serializer {
private:
actor_addressing* m_addressing;
actor_namespace* m_namespace;
type_lookup_table* m_outgoing_types;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include"cppa/actor_addressing.hpp"
namespace cppa {
actor_addressing::~actor_addressing() { }
} // namespace cppa
......@@ -25,35 +25,26 @@
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
\******************************************************************************/
#include <cstdint>
#include <utility>
#include "cppa/logging.hpp"
#include "cppa/to_string.hpp"
#include "cppa/serializer.hpp"
#include "cppa/singletons.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/process_information.hpp"
#include "cppa/io/default_actor_proxy.hpp"
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
using namespace std;
namespace cppa {
namespace cppa { namespace io {
default_actor_addressing::default_actor_addressing(default_protocol* parent)
: m_parent(parent), m_pinf(process_information::get()) { }
atom_value default_actor_addressing::technology_id() const {
return atom("DEFAULT");
}
void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
void actor_namespace::write(serializer* sink, const actor_ptr& ptr) {
CPPA_REQUIRE(sink != nullptr);
if (ptr == nullptr) {
CPPA_LOG_DEBUG("serialize nullptr");
......@@ -65,9 +56,9 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
if (!ptr->is_proxy()) {
get_actor_registry()->put(ptr->id(), ptr);
}
auto pinf = m_pinf;
auto pinf = process_information::get();
if (ptr->is_proxy()) {
auto dptr = ptr.downcast<default_actor_proxy>();
auto dptr = ptr.downcast<io::remote_actor_proxy>();
if (dptr) pinf = dptr->process_info();
else CPPA_LOG_ERROR("downcast failed");
}
......@@ -77,8 +68,8 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
pinf->node_id().data());
}
}
actor_ptr default_actor_addressing::read(deserializer* source) {
actor_ptr actor_namespace::read(deserializer* source) {
CPPA_REQUIRE(source != nullptr);
process_information::node_id_type nid;
auto aid = source->read<uint32_t>();
......@@ -93,82 +84,78 @@ actor_ptr default_actor_addressing::read(deserializer* source) {
return get_actor_registry()->get(aid);
}
else {
process_information tmp{pid, nid};
process_information_ptr tmp = new process_information{pid, nid};
return get_or_put(tmp, aid);
}
return nullptr;
}
size_t default_actor_addressing::count_proxies(const process_information& inf) {
auto i = m_proxies.find(inf);
size_t actor_namespace::count_proxies(const process_information& node) {
auto i = m_proxies.find(node);
return (i != m_proxies.end()) ? i->second.size() : 0;
}
actor_ptr default_actor_addressing::get(const process_information& inf,
actor_id aid) {
auto& submap = m_proxies[inf];
actor_ptr actor_namespace::get(const process_information& node, actor_id aid) {
auto& submap = m_proxies[node];
auto i = submap.find(aid);
if (i != submap.end()) {
auto result = i->second.promote();
CPPA_LOGMF_IF(!result, CPPA_INFO, self, "proxy instance expired; "
<< CPPA_TARG(inf, to_string) << ", "<< CPPA_ARG(aid));
CPPA_LOG_INFO_IF(!result, "proxy instance expired; "
<< CPPA_TARG(node, to_string)
<< ", "<< CPPA_ARG(aid));
if (!result) submap.erase(i);
return result;
}
return nullptr;
}
actor_ptr actor_namespace::get_or_put(process_information_ptr node, actor_id aid) {
auto result = get(*node, aid);
if (result == nullptr && m_factory) {
auto ptr = m_factory(aid, node);
put(*node, aid, ptr);
result = ptr;
}
return result;
}
void default_actor_addressing::put(const process_information& node,
actor_id aid,
const actor_proxy_ptr& proxy) {
void actor_namespace::put(const process_information& node,
actor_id aid,
const actor_proxy_ptr& proxy) {
auto& submap = m_proxies[node];
auto i = submap.find(aid);
if (i == submap.end()) {
submap.insert(make_pair(aid, proxy));
m_parent->enqueue(node,
{nullptr, nullptr},
make_any_tuple(atom("MONITOR"),
process_information::get(),
aid));
submap.insert(std::make_pair(aid, proxy));
if (m_new_element_callback) m_new_element_callback(aid, node);
/*if (m_parent) {
m_parent->enqueue(node,
{nullptr, nullptr},
make_any_tuple(atom("MONITOR"),
process_information::get(),
aid));
}*/
}
else {
CPPA_LOGMF(CPPA_ERROR, self, "proxy for " << aid << ":"
<< to_string(node) << " already exists");
CPPA_LOG_ERROR("proxy for " << aid << ":"
<< to_string(node) << " already exists");
}
}
actor_ptr default_actor_addressing::get_or_put(const process_information& inf,
actor_id aid) {
auto result = get(inf, aid);
if (result == nullptr) {
CPPA_LOGMF(CPPA_INFO, self, "created new proxy instance; "
<< CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
if (m_parent == nullptr) {
CPPA_LOG_ERROR("m_parent == nullptr (cannot create proxy without MM)");
}
else {
auto ptr = make_counted<default_actor_proxy>(aid, new process_information(inf), m_parent);
put(inf, aid, ptr);
result = ptr;
}
}
return result;
auto actor_namespace::proxies(process_information& node) -> proxy_map& {
return m_proxies[node];
}
auto default_actor_addressing::proxies(process_information& i) -> proxy_map& {
return m_proxies[i];
}
void default_actor_addressing::erase(process_information& inf) {
void actor_namespace::erase(process_information& inf) {
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string));
m_proxies.erase(inf);
}
void default_actor_addressing::erase(process_information& inf, actor_id aid) {
void actor_namespace::erase(process_information& inf, actor_id aid) {
CPPA_LOGMF(CPPA_TRACE, self, CPPA_TARG(inf, to_string) << ", " << CPPA_ARG(aid));
auto i = m_proxies.find(inf);
if (i != m_proxies.end()) {
i->second.erase(aid);
}
}
} } // namespace cppa::network
} // namespace cppa
......@@ -50,8 +50,7 @@ typedef cppa::util::upgrade_lock_guard<cppa::util::shared_spinlock> upgrade_guar
namespace cppa { namespace detail {
actor_registry::actor_registry() : m_running(0), m_ids(1) {
}
actor_registry::actor_registry() : m_running(0), m_ids(1) { }
actor_registry::value_type actor_registry::get_entry(actor_id key) const {
shared_guard guard(m_instances_mtx);
......
......@@ -158,14 +158,14 @@ struct pt_reader {
} // namespace <anonmyous>
binary_deserializer::binary_deserializer(const void* buf, size_t buf_size,
actor_addressing* addressing,
actor_namespace* ns,
type_lookup_table* tbl)
: super(addressing, tbl), m_pos(buf), m_end(advanced(buf, buf_size)) { }
: super(ns, tbl), m_pos(buf), m_end(advanced(buf, buf_size)) { }
binary_deserializer::binary_deserializer(const void* bbegin, const void* bend,
actor_addressing* addressing,
actor_namespace* ns,
type_lookup_table* tbl)
: super(addressing, tbl), m_pos(bbegin), m_end(bend) { }
: super(ns, tbl), m_pos(bbegin), m_end(bend) { }
const uniform_type_info* binary_deserializer::begin_object() {
std::uint8_t flag;
......
......@@ -121,9 +121,9 @@ class binary_writer {
} // namespace <anonymous>
binary_serializer::binary_serializer(util::buffer* buf,
actor_addressing* aa,
actor_namespace* ns,
type_lookup_table* tbl)
: super(aa, tbl), m_sink(buf) { }
: super(ns, tbl), m_sink(buf) { }
void binary_serializer::begin_object(const uniform_type_info* uti) {
CPPA_REQUIRE(uti != nullptr);
......
......@@ -111,23 +111,23 @@ class broker::servant : public continuable {
template<typename... Ts>
servant(broker_ptr parent, Ts&&... args)
: super{std::forward<Ts>(args)...}, m_disconnected{false}
, m_parent{move(parent)} { }
, m_broker{move(parent)} { }
void io_failed(event_bitmask mask) override {
if (mask == event::read) disconnect();
}
void dispose() override {
parent().erase_io(read_handle());
if (parent().m_io.empty() && parent().m_accept.empty()) {
m_broker->erase_io(read_handle());
if (m_broker->m_io.empty() && m_broker->m_accept.empty()) {
// release implicit reference count held by middleman
// in caes no reader/writer is left for this broker
parent().deref();
m_broker->deref();
}
}
void set_parent(broker_ptr new_parent) {
if (!m_disconnected) m_parent = std::move(new_parent);
void set_broker(broker_ptr new_broker) {
if (!m_disconnected) m_broker = std::move(new_broker);
}
protected:
......@@ -135,21 +135,17 @@ class broker::servant : public continuable {
void disconnect() {
if (!m_disconnected) {
m_disconnected = true;
if (parent().exit_reason() == exit_reason::not_exited) {
parent().invoke_message(nullptr, disconnect_message());
if (m_broker->exit_reason() == exit_reason::not_exited) {
m_broker->invoke_message(nullptr, disconnect_message());
}
}
}
virtual any_tuple disconnect_message() = 0;
broker& parent() const { return *m_parent; }
bool m_disconnected;
private:
broker_ptr m_parent;
broker_ptr m_broker;
};
......@@ -184,7 +180,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
});
for (;;) {
// stop reading if actor finished execution
if (parent().exit_reason() != exit_reason::not_exited) {
if (m_broker->exit_reason() != exit_reason::not_exited) {
return read_closed;
}
auto& buf = get_ref<2>(m_read_msg);
......@@ -211,7 +207,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
|| m_policy == broker::exactly
|| m_policy == broker::at_most) {
CPPA_LOG_DEBUG("invoke io actor");
parent().invoke_message(nullptr, m_read_msg);
m_broker->invoke_message(nullptr, m_read_msg);
CPPA_LOG_INFO_IF(!m_read_msg.vals()->unique(), "detached buffer");
get_ref<2>(m_read_msg).clear();
}
......@@ -266,9 +262,9 @@ class broker::doorman : public broker::servant {
if (opt) {
using namespace std;
auto& p = *opt;
get_ref<2>(m_accept_msg) = parent().add_scribe(move(p.first),
move(p.second));
parent().invoke_message(nullptr, m_accept_msg);
get_ref<2>(m_accept_msg) = m_broker->add_scribe(move(p.first),
move(p.second));
m_broker->invoke_message(nullptr, m_accept_msg);
}
else return read_continue_later;
}
......@@ -465,7 +461,7 @@ actor_ptr broker::fork_impl(std::function<void (broker*)> fun,
scribe* sptr = i->second.get(); // non-owning pointer
auto result = make_counted<default_broker>(move(fun), move(i->second));
init_and_launch(result);
sptr->set_parent(result); // set new parent
sptr->set_broker(result); // set new broker
m_io.erase(i);
return result;
}
......
This diff is collapsed.
......@@ -40,8 +40,8 @@
namespace cppa {
deserializer::deserializer(actor_addressing* aa, type_lookup_table* ot)
: m_addressing{aa}, m_incoming_types{ot} { }
deserializer::deserializer(actor_namespace* ns, type_lookup_table* ot)
: m_namespace{ns}, m_incoming_types{ot} { }
deserializer::~deserializer() { }
......
......@@ -51,12 +51,14 @@
#include "cppa/util/buffer.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/peer.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/io/peer_acceptor.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
#include "cppa/io/default_message_queue.hpp"
#include "cppa/io/middleman_event_handler.hpp"
#include "cppa/detail/fd_util.hpp"
......@@ -86,62 +88,167 @@ class middleman_event {
};
void middleman::continue_writer(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->add_later(ptr, event::write);
}
void middleman::stop_writer(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->erase_later(ptr, event::write);
}
bool middleman::has_writer(continuable* ptr) {
return m_handler->has_writer(ptr);
}
void middleman::continue_reader(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->add_later(ptr, event::read);
}
void middleman::stop_reader(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->erase_later(ptr, event::read);
}
bool middleman::has_reader(continuable* ptr) {
return m_handler->has_reader(ptr);
}
typedef intrusive::single_reader_queue<middleman_event> middleman_queue;
class middleman_impl {
/*
* A middleman also implements a "namespace" for actors.
*/
class middleman_impl : public middleman {
friend class middleman;
friend void middleman_loop(middleman_impl*);
public:
middleman_impl(std::unique_ptr<protocol>&& proto)
: m_done(false), m_handler(middleman_event_handler::create())
, m_protocol(std::move(proto)) { }
protocol* get_protocol() {
return m_protocol.get();
middleman_impl() : m_done(false) {
m_handler = middleman_event_handler::create();
m_namespace.set_proxy_factory([=](actor_id aid, process_information_ptr ptr) {
return make_counted<remote_actor_proxy>(aid, std::move(ptr), this);
});
m_namespace.set_new_element_callback([=](actor_id aid, const process_information& node) {
deliver(node,
{nullptr, nullptr},
make_any_tuple(atom("MONITOR"),
process_information::get(),
aid));
});
}
void run_later(function<void()> fun) {
void run_later(function<void()> fun) override {
m_queue.enqueue(new middleman_event(move(fun)));
atomic_thread_fence(memory_order_seq_cst);
uint8_t dummy = 0;
// ignore result; write error only means middleman already exited
static_cast<void>(write(m_pipe_write, &dummy, sizeof(dummy)));
static_cast<void>(::write(m_pipe_write, &dummy, sizeof(dummy)));
}
void continue_writer(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->add_later(ptr, event::write);
void register_peer(const process_information& node, peer* ptr) override {
CPPA_LOG_TRACE("node = " << to_string(node) << ", ptr = " << ptr);
auto& entry = m_peers[node];
if (entry.impl == nullptr) {
if (entry.queue == nullptr) entry.queue.emplace();
ptr->set_queue(entry.queue);
entry.impl = ptr;
if (!entry.queue->empty()) {
auto tmp = entry.queue->pop();
ptr->enqueue(tmp.first, tmp.second);
}
CPPA_LOG_INFO("peer " << to_string(node) << " added");
}
else {
CPPA_LOG_WARNING("peer " << to_string(node) << " already defined, "
"multiple calls to remote_actor()?");
}
}
void stop_writer(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->erase_later(ptr, event::write);
peer* get_peer(const process_information& node) override {
CPPA_LOG_TRACE("n = " << to_string(n));
auto i = m_peers.find(node);
if (i != m_peers.end()) {
CPPA_LOG_DEBUG("result = " << i->second.impl);
return i->second.impl;
}
CPPA_LOGMF(CPPA_DEBUG, self, "result = nullptr");
return nullptr;
}
inline bool has_writer(continuable* ptr) {
return m_handler->has_writer(ptr);
void del_acceptor(peer_acceptor* ptr) override {
auto i = m_acceptors.begin();
auto e = m_acceptors.end();
while (i != e) {
auto& vec = i->second;
auto last = vec.end();
auto iter = std::find(vec.begin(), last, ptr);
if (iter != last) vec.erase(iter);
if (not vec.empty()) ++i;
else i = m_acceptors.erase(i);
}
}
void continue_reader(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->add_later(ptr, event::read);
void deliver(const process_information& node,
const message_header& hdr,
any_tuple msg ) override {
auto& entry = m_peers[node];
if (entry.impl) {
CPPA_REQUIRE(entry.queue != nullptr);
if (!entry.impl->has_unwritten_data()) {
CPPA_REQUIRE(entry.queue->empty());
entry.impl->enqueue(hdr, msg);
return;
}
}
if (entry.queue == nullptr) entry.queue.emplace();
entry.queue->emplace(hdr, msg);
}
void stop_reader(continuable* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_handler->erase_later(ptr, event::read);
void last_proxy_exited(peer* pptr) override {
CPPA_REQUIRE(pptr != nullptr);
CPPA_LOG_TRACE(CPPA_ARG(pptr)
<< ", pptr->node() = " << to_string(pptr->node()));
if (pptr->erase_on_last_proxy_exited() && pptr->queue().empty()) {
stop_reader(pptr);
auto i = m_peers.find(pptr->node());
if (i != m_peers.end()) {
CPPA_LOG_DEBUG_IF(i->second.impl != pptr,
"node " << to_string(pptr->node())
<< " does not exist in m_peers");
if (i->second.impl == pptr) {
m_peers.erase(i);
}
}
}
}
inline bool has_reader(continuable* ptr) {
return m_handler->has_reader(ptr);
void new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node = nullptr) override {
CPPA_LOG_TRACE("");
auto ptr = new peer(this, in, out, node);
continue_reader(ptr);
if (node) register_peer(*node, ptr);
}
void register_acceptor(const actor_ptr& whom, peer_acceptor* ptr) override {
run_later([=] {
CPPA_LOGC_TRACE("cppa::io::middleman",
"register_acceptor$lambda", "");
m_acceptors[whom].push_back(ptr);
continue_reader(ptr);
});
}
protected:
void initialize() {
void initialize() override {
int pipefds[2];
if (pipe(pipefds) != 0) { CPPA_CRITICAL("cannot create pipe"); }
m_pipe_read = pipefds[0];
......@@ -151,7 +258,7 @@ class middleman_impl {
m_thread = thread([this] { middleman_loop(this); });
}
void destroy() {
void destroy() override {
run_later([this] {
CPPA_LOGM_TRACE("destroy$helper", "");
this->m_done = true;
......@@ -175,22 +282,17 @@ class middleman_impl {
native_socket_type m_pipe_read;
native_socket_type m_pipe_write;
middleman_queue m_queue;
std::unique_ptr<middleman_event_handler> m_handler;
std::unique_ptr<protocol> m_protocol;
struct peer_entry {
peer* impl;
default_message_queue_ptr queue;
};
std::map<actor_ptr, std::vector<peer_acceptor*>> m_acceptors;
std::map<process_information, peer_entry> m_peers;
};
void middleman::set_pimpl(std::unique_ptr<protocol>&& proto) {
m_impl.reset(new middleman_impl(std::move(proto)));
}
middleman* middleman::create_singleton() {
auto ptr = new middleman;
ptr->set_pimpl(std::unique_ptr<protocol>{new default_protocol(ptr)});
return ptr;
}
class middleman_overseer : public continuable {
typedef continuable super;
......@@ -244,47 +346,6 @@ class middleman_overseer : public continuable {
middleman::~middleman() { }
void middleman::destroy() {
m_impl->destroy();
}
void middleman::initialize() {
m_impl->initialize();
}
protocol* middleman::get_protocol() {
return m_impl->get_protocol();
}
void middleman::run_later(std::function<void()> fun) {
m_impl->run_later(std::move(fun));
}
void middleman::continue_writer(continuable* ptr) {
m_impl->continue_writer(ptr);
}
void middleman::stop_writer(continuable* ptr) {
m_impl->stop_writer(ptr);
}
bool middleman::has_writer(continuable* ptr) {
return m_impl->has_writer(ptr);
}
void middleman::continue_reader(continuable* ptr) {
m_impl->continue_reader(ptr);
}
void middleman::stop_reader(continuable* ptr) {
m_impl->stop_reader(ptr);
}
bool middleman::has_reader(continuable* ptr) {
return m_impl->has_reader(ptr);
}
void middleman_loop(middleman_impl* impl) {
# ifdef CPPA_LOG_LEVEL
auto mself = make_counted<thread_mapped_actor>();
......@@ -387,4 +448,8 @@ void middleman_loop(middleman_impl* impl) {
CPPA_LOGF_DEBUG("middleman loop done");
}
middleman* middleman::create_singleton() {
return new middleman_impl;
}
} } // namespace cppa::detail
......@@ -51,21 +51,19 @@
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/io/peer.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/default_protocol.hpp"
using namespace std;
namespace cppa { namespace io {
default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr)
: super(parent->parent(), out, in->read_handle(), out->write_handle())
, m_parent(parent), m_in(in)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
peer::peer(middleman* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr)
: super(parent, out, in->read_handle(), out->write_handle())
, m_in(in), m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_node(peer_ptr) {
m_rd_buf.final_size( m_state == wait_for_process_info
? sizeof(uint32_t) + process_information::node_id_size
......@@ -77,24 +75,24 @@ default_peer::default_peer(default_protocol* parent,
m_meta_msg = uniform_typeid<any_tuple>();
}
void default_peer::io_failed(event_bitmask mask) {
void peer::io_failed(event_bitmask mask) {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr")
<< " mask = " << mask);
// make sure this code is executed only once by filtering for read failure
if (mask == event::read && m_node) {
// kill all proxies
auto& children = m_parent->addressing()->proxies(*m_node);
auto& children = parent()->get_namespace().proxies(*m_node);
for (auto& kvp : children) {
auto ptr = kvp.second.promote();
if (ptr) ptr->enqueue(nullptr,
make_any_tuple(atom("KILL_PROXY"),
exit_reason::remote_link_unreachable));
}
m_parent->addressing()->erase(*m_node);
parent()->get_namespace().erase(*m_node);
}
}
continue_reading_result default_peer::continue_reading() {
continue_reading_result peer::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
try { m_rd_buf.append_from(m_in.get()); }
......@@ -119,7 +117,7 @@ continue_reading_result default_peer::continue_reading() {
return read_failure;
}
CPPA_LOG_DEBUG("read process info: " << to_string(*m_node));
m_parent->register_peer(*m_node, this);
parent()->register_peer(*m_node, this);
// initialization done
m_state = wait_for_msg_size;
m_rd_buf.clear();
......@@ -140,7 +138,7 @@ continue_reading_result default_peer::continue_reading() {
message_header hdr;
any_tuple msg;
binary_deserializer bd(m_rd_buf.data(), m_rd_buf.size(),
m_parent->addressing(), &m_incoming_types);
&(parent()->get_namespace()), &m_incoming_types);
try {
m_meta_hdr->deserialize(&hdr, &bd);
m_meta_msg->deserialize(&msg, &bd);
......@@ -190,7 +188,7 @@ continue_reading_result default_peer::continue_reading() {
}
}
void default_peer::monitor(const actor_ptr&,
void peer::monitor(const actor_ptr&,
const process_information_ptr& node,
actor_id aid) {
CPPA_LOG_TRACE(CPPA_MARG(node, get) << ", " << CPPA_ARG(aid));
......@@ -221,20 +219,20 @@ void default_peer::monitor(const actor_ptr&,
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "attach functor to " << entry.first.get());
default_protocol* proto = m_parent;
auto mm = parent();
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
CPPA_LOGC_TRACE("cppa::io::default_peer",
mm->run_later([=] {
CPPA_LOGC_TRACE("cppa::io::peer",
"monitor$kill_proxy_helper",
"reason = " << reason);
auto p = proto->get_peer(*node);
auto p = mm->get_peer(*node);
if (p) p->enqueue(make_any_tuple(atom("KILL_PROXY"), pself, aid, reason));
});
});
}
}
void default_peer::kill_proxy(const actor_ptr& sender,
void peer::kill_proxy(const actor_ptr& sender,
const process_information_ptr& node,
actor_id aid,
std::uint32_t reason) {
......@@ -250,7 +248,7 @@ void default_peer::kill_proxy(const actor_ptr& sender,
CPPA_LOGMF(CPPA_ERROR, self, "sender != nullptr");
return;
}
auto proxy = m_parent->addressing()->get(*node, aid);
auto proxy = parent()->get_namespace().get(*node, aid);
if (proxy) {
CPPA_LOGMF(CPPA_DEBUG, self, "received KILL_PROXY for " << aid
<< ":" << to_string(*node));
......@@ -263,7 +261,7 @@ void default_peer::kill_proxy(const actor_ptr& sender,
}
}
void default_peer::deliver(const message_header& hdr, any_tuple msg) {
void peer::deliver(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE("");
if (hdr.sender && hdr.sender->is_proxy()) {
hdr.sender.downcast<actor_proxy>()->deliver(hdr, std::move(msg));
......@@ -271,7 +269,7 @@ void default_peer::deliver(const message_header& hdr, any_tuple msg) {
else hdr.deliver(std::move(msg));
}
void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
void peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
// this message is sent from default_actor_proxy in link_to and
// establish_backling to cause the original actor (sender) to establish
// a link to ptr as well
......@@ -292,7 +290,7 @@ void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
}
}
void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
void peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
CPPA_LOG_TRACE(CPPA_MARG(sender, get)
<< ", " << CPPA_MARG(ptr, get));
CPPA_LOG_ERROR_IF(!sender, "received 'UNLINK' from invalid sender");
......@@ -307,7 +305,7 @@ void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
else sender->unlink_from(ptr);
}
continue_writing_result default_peer::continue_writing() {
continue_writing_result peer::continue_writing() {
CPPA_LOG_TRACE("");
auto result = super::continue_writing();
while (result == write_done && !queue().empty()) {
......@@ -316,14 +314,14 @@ continue_writing_result default_peer::continue_writing() {
result = super::continue_writing();
}
if (result == write_done && erase_on_last_proxy_exited() && !has_unwritten_data()) {
if (m_parent->addressing()->count_proxies(*m_node) == 0) {
m_parent->last_proxy_exited(this);
if (parent()->get_namespace().count_proxies(*m_node) == 0) {
parent()->last_proxy_exited(this);
}
}
return result;
}
void default_peer::add_type_if_needed(const std::string& tname) {
void peer::add_type_if_needed(const std::string& tname) {
if (m_outgoing_types.id_of(tname) == 0) {
auto id = m_outgoing_types.max_id() + 1;
auto imap = get_uniform_type_info_map();
......@@ -333,19 +331,19 @@ void default_peer::add_type_if_needed(const std::string& tname) {
}
}
void default_peer::enqueue_impl(const message_header& hdr, const any_tuple& msg) {
void peer::enqueue_impl(const message_header& hdr, const any_tuple& msg) {
CPPA_LOG_TRACE("");
auto tname = msg.tuple_type_names();
add_type_if_needed((tname) ? *tname : detail::get_tuple_type_names(*msg.vals()));
uint32_t size = 0;
auto& wbuf = write_buffer();
auto before = wbuf.size();
binary_serializer bs(&wbuf, m_parent->addressing(), &m_outgoing_types);
binary_serializer bs(&wbuf, &(parent()->get_namespace()), &m_outgoing_types);
wbuf.write(sizeof(uint32_t), &size);
try { bs << hdr << msg; }
catch (exception& e) {
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e));
cerr << "*** exception in default_peer::enqueue; "
cerr << "*** exception in peer::enqueue; "
<< to_verbose_string(e)
<< endl;
return;
......@@ -356,13 +354,13 @@ void default_peer::enqueue_impl(const message_header& hdr, const any_tuple& msg)
memcpy(wbuf.offset_data(before), &size, sizeof(std::uint32_t));
}
void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
void peer::enqueue(const message_header& hdr, const any_tuple& msg) {
enqueue_impl(hdr, msg);
register_for_writing();
}
void default_peer::dispose() {
m_parent->del_peer(this);
void peer::dispose() {
parent()->get_namespace().erase(*m_node);
delete this;
}
......
......@@ -35,9 +35,8 @@
#include "cppa/to_string.hpp"
#include "cppa/process_information.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/default_peer_acceptor.hpp"
#include "cppa/io/peer.hpp"
#include "cppa/io/peer_acceptor.hpp"
#include "cppa/detail/demangle.hpp"
......@@ -45,12 +44,12 @@ using namespace std;
namespace cppa { namespace io {
default_peer_acceptor::default_peer_acceptor(default_protocol* parent,
acceptor_uptr aur,
const actor_ptr& pa)
peer_acceptor::peer_acceptor(middleman* parent,
acceptor_uptr aur,
const actor_ptr& pa)
: super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur)), m_pa(pa) { }
continue_reading_result default_peer_acceptor::continue_reading() {
continue_reading_result peer_acceptor::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
optional<stream_ptr_pair> opt{none};
......@@ -83,12 +82,12 @@ continue_reading_result default_peer_acceptor::continue_reading() {
}
}
void default_peer_acceptor::io_failed(event_bitmask) {
CPPA_LOG_INFO("removed default_peer_acceptor "
void peer_acceptor::io_failed(event_bitmask) {
CPPA_LOG_INFO("removed peer_acceptor "
<< this << " due to an IO failure");
}
void default_peer_acceptor::dispose() {
void peer_acceptor::dispose() {
m_parent->del_acceptor(this);
delete this;
}
......
......@@ -31,8 +31,10 @@
#include "cppa/to_string.hpp"
#include "cppa/logging.hpp"
#include "cppa/io/peer.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_actor_proxy.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -49,37 +51,37 @@ inline sync_request_info* new_req_info(actor_ptr sptr, message_id id) {
sync_request_info::sync_request_info(actor_ptr sptr, message_id id)
: next(nullptr), sender(std::move(sptr)), mid(id) { }
default_actor_proxy::default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
default_protocol* parent)
remote_actor_proxy::remote_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
middleman* parent)
: super(mid), m_parent(parent), m_pinf(pinfo) {
CPPA_REQUIRE(parent != nullptr);
CPPA_LOG_INFO(CPPA_ARG(mid) << ", " << CPPA_TARG(pinfo, to_string)
<< "protocol = " << detail::demangle(typeid(*parent)));
}
default_actor_proxy::~default_actor_proxy() {
remote_actor_proxy::~remote_actor_proxy() {
auto aid = m_id;
auto node = m_pinf;
auto proto = m_parent;
auto mm = m_parent;
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(m_pinf)
<< ", protocol = " << detail::demangle(typeid(*m_parent)));
proto->run_later([aid, node, proto] {
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
"~default_actor_proxy$run_later",
mm->run_later([aid, node, mm] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"~remote_actor_proxy$run_later",
"node = " << to_string(*node) << ", aid " << aid
<< ", proto = " << to_string(proto->identifier()));
proto->addressing()->erase(*node, aid);
auto p = proto->get_peer(*node);
mm->get_namespace().erase(*node, aid);
auto p = mm->get_peer(*node);
if (p && p->erase_on_last_proxy_exited()) {
if (proto->addressing()->count_proxies(*node) == 0) {
proto->last_proxy_exited(p);
if (mm->get_namespace().count_proxies(*node) == 0) {
mm->last_proxy_exited(p);
}
}
});
}
void default_actor_proxy::deliver(const message_header& hdr, any_tuple msg) {
void remote_actor_proxy::deliver(const message_header& hdr, any_tuple msg) {
// this member function is exclusively called from default_peer from inside
// the middleman's thread, therefore we can safely access
// m_pending_requests here
......@@ -93,7 +95,7 @@ void default_actor_proxy::deliver(const message_header& hdr, any_tuple msg) {
hdr.deliver(std::move(msg));
}
void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg) {
void remote_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_ARG(m_id) << ", " << CPPA_TSARG(hdr)
<< ", " << CPPA_TSARG(msg));
if (hdr.receiver != this) {
......@@ -107,7 +109,7 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
case intrusive::queue_closed: {
auto rsn = exit_reason();
m_parent->run_later([rsn, hdr] {
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"forward_msg$bouncer",
"bounce message for reason " << rsn);
detail::sync_request_bouncer f{rsn};
......@@ -119,16 +121,16 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
}
}
auto node = m_pinf;
auto proto = m_parent;
m_parent->run_later([hdr, msg, node, proto] {
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
auto mm = m_parent;
m_parent->run_later([hdr, msg, node, mm] {
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"forward_msg$forwarder",
"");
proto->enqueue(*node, hdr, msg);
mm->deliver(*node, hdr, msg);
});
}
void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
void remote_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_TARG(hdr, to_string) << ", " << CPPA_TARG(msg, to_string));
auto& arr = detail::static_types_array<atom_value, uint32_t>::arr;
if ( msg.size() == 2
......@@ -136,10 +138,10 @@ void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
&& msg.get_as<atom_value>(0) == atom("KILL_PROXY")
&& msg.type_at(1) == arr[1]) {
CPPA_LOG_DEBUG("received KILL_PROXY message");
intrusive_ptr<default_actor_proxy> _this{this};
intrusive_ptr<remote_actor_proxy> _this{this};
auto reason = msg.get_as<uint32_t>(1);
m_parent->run_later([_this, reason] {
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::remote_actor_proxy",
"enqueue$kill_proxy_helper",
"KILL_PROXY with exit reason " << reason);
_this->cleanup(reason);
......@@ -152,7 +154,7 @@ void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
else forward_msg(hdr, move(msg));
}
void default_actor_proxy::link_to(const intrusive_ptr<actor>& other) {
void remote_actor_proxy::link_to(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
if (link_to_impl(other)) {
// causes remote actor to link to (proxy of) other
......@@ -161,7 +163,7 @@ void default_actor_proxy::link_to(const intrusive_ptr<actor>& other) {
}
}
void default_actor_proxy::unlink_from(const intrusive_ptr<actor>& other) {
void remote_actor_proxy::unlink_from(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
if (unlink_from_impl(other)) {
// causes remote actor to unlink from (proxy of) other
......@@ -169,7 +171,7 @@ void default_actor_proxy::unlink_from(const intrusive_ptr<actor>& other) {
}
}
bool default_actor_proxy::establish_backlink(const intrusive_ptr<actor>& other) {
bool remote_actor_proxy::establish_backlink(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
if (super::establish_backlink(other)) {
// causes remote actor to unlink from (proxy of) other
......@@ -179,7 +181,7 @@ bool default_actor_proxy::establish_backlink(const intrusive_ptr<actor>& other)
return false;
}
bool default_actor_proxy::remove_backlink(const intrusive_ptr<actor>& other) {
bool remote_actor_proxy::remove_backlink(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
if (super::remove_backlink(other)) {
// causes remote actor to unlink from (proxy of) other
......@@ -189,12 +191,12 @@ bool default_actor_proxy::remove_backlink(const intrusive_ptr<actor>& other) {
return false;
}
void default_actor_proxy::local_link_to(const intrusive_ptr<actor>& other) {
void remote_actor_proxy::local_link_to(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
link_to_impl(other);
}
void default_actor_proxy::local_unlink_from(const intrusive_ptr<actor>& other) {
void remote_actor_proxy::local_unlink_from(const intrusive_ptr<actor>& other) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
unlink_from_impl(other);
}
......
......@@ -32,8 +32,8 @@
namespace cppa {
serializer::serializer(actor_addressing* aa, type_lookup_table* it)
: m_addressing{aa}, m_outgoing_types{it} { }
serializer::serializer(actor_namespace* ns, type_lookup_table* it)
: m_namespace{ns}, m_outgoing_types{it} { }
serializer::~serializer() { }
......
......@@ -43,13 +43,12 @@
#include "cppa/singletons.hpp"
#include "cppa/from_string.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/util/algorithm.hpp"
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
using namespace std;
......@@ -70,7 +69,7 @@ class string_serializer : public serializer {
typedef serializer super;
ostream& out;
io::default_actor_addressing m_addressing;
actor_namespace m_namespace;
struct pt_writer {
......@@ -124,7 +123,7 @@ class string_serializer : public serializer {
public:
string_serializer(ostream& mout)
: super(&m_addressing), out(mout), m_after_value(false)
: super(&m_namespace), out(mout), m_after_value(false)
, m_obj_just_opened(false) { }
void begin_object(const uniform_type_info* uti) {
......@@ -207,7 +206,7 @@ class string_deserializer : public deserializer {
//size_t m_obj_count;
stack<bool> m_obj_had_left_parenthesis;
stack<string> m_open_objects;
io::default_actor_addressing m_addressing;
actor_namespace m_namespace;
void skip_space_and_comma() {
while (*m_pos == ' ' || *m_pos == ',') ++m_pos;
......@@ -275,7 +274,7 @@ class string_deserializer : public deserializer {
public:
string_deserializer(string str) : super(&m_addressing), m_str(move(str)) {
string_deserializer(string str) : super(&m_namespace), m_str(move(str)) {
m_pos = m_str.begin();
}
......
......@@ -49,37 +49,74 @@
#include "cppa/binary_deserializer.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/intrusive/blocking_single_reader_queue.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/peer_acceptor.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "cppa/io/remote_actor_proxy.hpp"
namespace cppa {
using namespace detail;
using namespace io;
namespace { protocol* proto() {
return get_middleman()->get_protocol();
} }
void publish(actor_ptr whom, std::unique_ptr<acceptor> aptr) {
proto()->publish(whom, move(aptr), {});
}
actor_ptr remote_actor(stream_ptr_pair io) {
return proto()->remote_actor(io, {});
CPPA_LOG_TRACE(CPPA_TARG(whom, to_string) << ", " << CPPA_MARG(ptr, get)
<< ", args.size() = " << args.size());
if (!whom) return;
CPPA_REQUIRE(args.size() == 0);
get_actor_registry()->put(whom->id(), whom);
auto mm = get_middleman();
mm->register_acceptor(whom, new peer_acceptor(mm, move(aptr), whom));
}
void publish(actor_ptr whom, std::uint16_t port, const char* addr) {
if (!addr) proto()->publish(whom, {port});
else proto()->publish(whom, {port, addr});
if (!whom) return;
publish(whom, ipv4_acceptor::create(port, addr));
}
actor_ptr remote_actor(stream_ptr_pair io) {
CPPA_LOG_TRACE("io{" << io.first.get() << ", " << io.second.get() << "}");
auto pinf = process_information::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->node_id().data(), pinf->node_id().size());
actor_id remote_aid;
std::uint32_t peer_pid;
process_information::node_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());
auto pinfptr = make_counted<process_information>(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");
return get_actor_registry()->get(remote_aid);
}
auto mm = get_middleman();
struct remote_actor_result { remote_actor_result* next; actor_ptr value; };
intrusive::blocking_single_reader_queue<remote_actor_result> q;
mm->run_later([mm, io, pinfptr, remote_aid, &q] {
CPPA_LOGC_TRACE("cppa",
"remote_actor$create_connection", "");
auto pp = mm->get_peer(*pinfptr);
CPPA_LOGF_INFO_IF(pp, "connection already exists (re-use old one)");
if (!pp) mm->new_peer(io.first, io.second, pinfptr);
auto res = mm->get_namespace().get_or_put(pinfptr, remote_aid);
q.push_back(new remote_actor_result{0, res});
});
std::unique_ptr<remote_actor_result> result(q.pop());
CPPA_LOGF_DEBUG("result = " << result->value.get());
return result->value;
}
actor_ptr remote_actor(const char* host, std::uint16_t port) {
return proto()->remote_actor({port, host});
auto io = ipv4_io_stream::connect_to(host, port);
return remote_actor(stream_ptr_pair(io, io));
}
} // namespace cppa
......@@ -48,7 +48,6 @@
#include "cppa/announce.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/actor_addressing.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/util/duration.hpp"
......
......@@ -39,7 +39,7 @@
#include "cppa/announce.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_header.hpp"
#include "cppa/actor_addressing.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/util/algorithm.hpp"
......@@ -164,15 +164,15 @@ inline void serialize_impl(const unit_t&, serializer*) { }
inline void deserialize_impl(unit_t&, deserializer*) { }
void serialize_impl(const actor_ptr& ptr, serializer* sink) {
auto impl = sink->addressing();
if (impl) impl->write(sink, ptr);
auto ns = sink->get_namespace();
if (ns) ns->write(sink, ptr);
else throw std::runtime_error("unable to serialize actor_ptr: "
"no actor addressing defined");
}
void deserialize_impl(actor_ptr& ptrref, deserializer* source) {
auto impl = source->addressing();
if (impl) ptrref = impl->read(source);
auto ns = source->get_namespace();
if (ns) ptrref = ns->read(source);
else throw std::runtime_error("unable to deserialize actor_ptr: "
"no actor addressing defined");
}
......
......@@ -38,6 +38,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/primitive_type.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
......@@ -49,8 +50,6 @@
#include "cppa/util/type_traits.hpp"
#include "cppa/util/abstract_uniform_type_info.hpp"
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/detail/ieee_754.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/type_to_ptype.hpp"
......@@ -169,7 +168,7 @@ int main() {
announce(typeid(raw_struct), create_unique<raw_struct_type_info>());
io::default_actor_addressing addressing;
actor_namespace addressing;
cout << "process id: " << to_string(process_information::get()) << endl;
......
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