Commit 561b43b0 authored by Dominik Charousset's avatar Dominik Charousset

implemented logging and default_protocol

libcppa emits log4j compatible output now to make use of available tools;
middleman uses `continuable_writer` and `continuable_reader` interfaces only;
´default_protocol´ encapsulates libcppa's binary protocol now;
`remote_actor` checks wheter a previous connection to the host already exists;
actor proxies are stored as weak pointers in peers, fixes #75;
`default_peer` created with `remote_actor` close connection on last proxy exit;
poll and epoll share meta data implementation and event-interface;
no global proxy cache, belongs to `protocol`/`actor_addressing` now;
new configure option: `--with-cppa-log-level`;
`to_string` is no longer a template (enables proper overload resolution)
parent 88953ad6
......@@ -65,6 +65,12 @@ if (ENABLE_DEBUG)
# set(CMAKE_BUILD_TYPE RelWithDebInfo)
endif ()
if (CPPA_LOG_LEVEL)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DCPPA_LOG_LEVEL=${CPPA_LOG_LEVEL}")
else ()
set(CPPA_LOG_LEVEL "0")
endif()
# set build default build type if not set
if (CMAKE_BUILD_TYPE)
else (CMAKE_BUILD_TYPE)
......@@ -207,6 +213,7 @@ message("\n====================| Build Summary |===================="
"\n"
"\nBuild type: ${CMAKE_BUILD_TYPE}"
"\nDebug mode: ${ENABLE_DEBUG}"
"\nLog level: ${CPPA_LOG_LEVEL}"
"\nContext switching: ${CONTEXT_SWITCHING}"
"\n"
"\nCXX: ${CMAKE_CXX_COMPILER}"
......
......@@ -37,6 +37,12 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--enable-debug compile in debugging mode
(always sets --build-type=Debug)
--enable-perftools use Google perftools
--with-cppa-log-level=LVL sets the debugging output, possible values:
- WARNING
- INFO
- DEBUG
- TRACE
(implicitly sets --enable-debug)
Platform-Dependent Adjustments:
--disable-context-switching compile libcppa without context-switching actors
......@@ -147,6 +153,28 @@ while [ $# -ne 0 ]; do
--enable-debug)
append_cache_entry ENABLE_DEBUG BOOL true
;;
--with-cppa-log-level=*)
level=$(echo "$optarg" | tr '[:lower:]' '[:upper:]')
case $level in
WARNING)
append_cache_entry CPPA_LOG_LEVEL STRING 1
;;
INFO)
append_cache_entry CPPA_LOG_LEVEL STRING 2
;;
DEBUG)
append_cache_entry CPPA_LOG_LEVEL STRING 3
;;
TRACE)
append_cache_entry CPPA_LOG_LEVEL STRING 4
;;
*)
echo "Invalid log level '$level'. Try '$0 --help' to see valid values."
exit 1
;;
esac
append_cache_entry ENABLE_DEBUG BOOL true
;;
--disable-context-switching)
append_cache_entry DISABLE_CONTEXT_SWITCHING BOOL true
;;
......
......@@ -38,7 +38,7 @@ cppa/deserializer.hpp
cppa/detail/abstract_scheduled_actor.hpp
cppa/detail/abstract_tuple.hpp
cppa/detail/actor_count.hpp
cppa/detail/actor_proxy_cache.hpp
cppa/actor_addressing.hpp
cppa/detail/actor_registry.hpp
cppa/network/addressed_message.hpp
cppa/detail/atom_val.hpp
......@@ -174,7 +174,6 @@ src/abstract_tuple.cpp
src/actor.cpp
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_proxy_cache.cpp
src/actor_registry.cpp
src/addressed_message.cpp
src/any_tuple.cpp
......@@ -269,16 +268,24 @@ cppa/detail/opt_impls.hpp
examples/type_plugins.hpp
cppa/detail/decorated_names_map.hpp
src/decorated_names_map.cpp
cppa/network/peer_acceptor.hpp
src/peer_acceptor.cpp
cppa/network/peer.hpp
src/peer.cpp
cppa/network/continuable_reader.hpp
src/continuable_reader.cpp
cppa/network/default_peer_impl.hpp
src/default_peer_impl.cpp
cppa/network/default_peer_acceptor_impl.hpp
src/default_peer_acceptor_impl.cpp
cppa/network/default_peer.hpp
src/default_peer.cpp
cppa/network/default_peer_acceptor.hpp
src/default_peer_acceptor.cpp
src/ref_counted.cpp
cppa/enable_weak_ptr_mixin.hpp
cppa/weak_intrusive_ptr.hpp
cppa/network/default_actor_proxy.hpp
src/default_actor_proxy.cpp
cppa/network/default_actor_addressing.hpp
src/default_actor_addressing.cpp
cppa/network/protocol.hpp
src/protocol.cpp
cppa/network/default_protocol.hpp
src/default_protocol.cpp
cppa/detail/logging.hpp
src/logging.cpp
cppa/network/continuable_writer.hpp
src/continuable_writer.cpp
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 ref_counted {
public:
/**
* @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.
* @note Implementation should call {@link actor_registry::put()}
* to be able to restore instances later on from registry.
* @note Thi
*/
virtual void write(serializer* sink, const actor_ptr& ptr) = 0;
/**
* @brief Deserializes an actor from @p source according
* to the implemented addressing.
*/
virtual actor_ptr read(deserializer* source) = 0;
};
} // namespace cppa::detail
#endif // CPPA_ACTOR_PROXY_CACHE_HPP
......@@ -32,57 +32,52 @@
#define CPPA_ACTOR_PROXY_HPP
#include "cppa/actor.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/weak_intrusive_ptr.hpp"
#include "cppa/enable_weak_ptr_mixin.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
class actor_proxy_cache;
/**
* @brief Represents a remote actor.
*/
class actor_proxy : public actor { };
class actor_proxy : public enable_weak_ptr_mixin<actor_proxy,actor> {
#else // CPPA_DOCUMENTATION
class actor_proxy : public detail::abstract_actor<actor> {
typedef abstract_actor<actor> super;
typedef enable_weak_ptr_mixin<actor_proxy,actor> super;
public:
actor_proxy(std::uint32_t mid, const process_information_ptr& parent);
void enqueue(actor* sender, any_tuple msg);
void sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
void link_to(const intrusive_ptr<actor>& other);
// do not cause to send this actor an "UNLINK" message
// to the "original" remote actor
void local_link_to(const intrusive_ptr<actor>& other);
/**
* @brief Establishes a local link state that's not synchronized back
* to the remote instance.
*/
virtual void local_link_to(const intrusive_ptr<actor>& other) = 0;
void unlink_from(const intrusive_ptr<actor>& other);
/**
* @brief Removes a local link state.
*/
virtual void local_unlink_from(const actor_ptr& other) = 0;
// do not cause to send this actor an "UNLINK" message
// to the "original" remote actor
void local_unlink_from(const actor_ptr& other);
protected:
bool remove_backlink(const intrusive_ptr<actor>& to);
bool establish_backlink(const intrusive_ptr<actor>& to);
actor_proxy(actor_id mid,
const process_information_ptr& pinfo);
};
#endif // CPPA_DOCUMENTATION
/**
* @brief A smart pointer to an {@link actor_proxy} instance.
* @relates actor_proxy
*/
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
/**
* @brief A weak smart pointer to an {@link actor_proxy} instance.
* @relates actor_proxy
*/
typedef weak_intrusive_ptr<actor_proxy> weak_actor_proxy_ptr;
} // namespace cppa
#endif // CPPA_ACTOR_PROXY_HPP
......@@ -41,13 +41,15 @@ namespace cppa {
*/
class binary_deserializer : public deserializer {
const char* pos;
const char* end;
typedef deserializer super;
public:
binary_deserializer(const char* buf, size_t buf_size);
binary_deserializer(const char* begin, const char* end);
binary_deserializer(const char* buf, size_t buf_size,
actor_addressing* addressing = nullptr);
binary_deserializer(const char* begin, const char* end,
actor_addressing* addressing = nullptr);
std::string seek_object();
std::string peek_object();
......@@ -61,6 +63,11 @@ class binary_deserializer : public deserializer {
primitive_variant* storage);
void read_raw(size_t num_bytes, void* storage);
private:
const char* pos;
const char* end;
};
} // namespace cppa
......
......@@ -46,13 +46,15 @@ namespace detail { class binary_writer; }
*/
class binary_serializer : public serializer {
typedef serializer super;
public:
/**
* @brief Creates a binary serializer writing to @p write_buffer.
* @warning @p write_buffer must be guaranteed to outlive @p this
*/
binary_serializer(util::buffer* write_buffer);
binary_serializer(util::buffer* write_buffer, actor_addressing* ptr = 0);
void begin_object(const std::string& tname);
......
......@@ -40,6 +40,7 @@
namespace cppa {
class object;
class actor_addressing;
/**
* @ingroup TypeSystem
......@@ -52,7 +53,7 @@ class deserializer {
public:
deserializer() = default;
deserializer(actor_addressing* addressing = nullptr);
virtual ~deserializer();
......@@ -113,6 +114,12 @@ class deserializer {
*/
virtual void read_raw(size_t num_bytes, void* storage) = 0;
inline actor_addressing* addressing() { return m_addressing; }
private:
actor_addressing* m_addressing;
};
/**
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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_LOGGING_HPP
#define CPPA_LOGGING_HPP
#include <sstream>
#include "cppa/detail/demangle.hpp"
/*
* To enable logging, you have to define CPPA_DEBUG. This enables
* CPPA_LOG_ERROR messages. To enable more debugging output, you can define
* CPPA_LOG_LEVEL to:
* 1: + warning
* 2: + info
* 3: + debug
* 4: + trace (prints for each logged method entry and exit message)
*
* Note: this logger emits log4j style XML output; logs are best viewed
* using a log4j viewer, e.g., http://code.google.com/p/otroslogviewer/
*
*/
namespace cppa { namespace detail {
class logging {
public:
virtual ~logging();
virtual void log(const char* level,
const char* class_name,
const char* function_name,
const char* file_name,
int line_num,
const std::string& msg ) = 0;
virtual void start() = 0;
virtual void stop() = 0;
static logging* instance();
static logging* create_singleton();
class trace_helper {
public:
inline trace_helper(std::string class_name,
const char* fun_name,
const char* file_name,
int line_num,
const std::string& msg)
: m_class(std::move(class_name)), m_fun_name(fun_name)
, m_file_name(file_name), m_line_num(line_num) {
logging::instance()->log("TRACE ", m_class.c_str(), fun_name,
file_name, line_num, "ENTRY " + msg);
}
inline ~trace_helper() {
logging::instance()->log("TRACE ", m_class.c_str(), m_fun_name,
m_file_name, m_line_num, "EXIT");
}
private:
std::string m_class;
const char* m_fun_name;
const char* m_file_name;
int m_line_num;
};
};
} } // namespace cppa::detail
#ifndef CPPA_DEBUG
#define CPPA_LOG_ERROR(unused) static_cast<void>(0)
#define CPPA_LOG_ERROR_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOGF_ERROR(unused) static_cast<void>(0)
#define CPPA_LOGF_ERROR_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOG_WARNING(unused) static_cast<void>(0)
#define CPPA_LOG_WARNING_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOGF_WARNING(unused) static_cast<void>(0)
#define CPPA_LOGF_WARNING_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOG_INFO(unused) static_cast<void>(0)
#define CPPA_LOG_INFO_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOGF_INFO(unused) static_cast<void>(0)
#define CPPA_LOGF_INFO_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOG_DEBUG(unused) static_cast<void>(0)
#define CPPA_LOG_DEBUG_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOGF_DEBUG(unused) static_cast<void>(0)
#define CPPA_LOGF_DEBUG_IF(unused1,unused2) static_cast<void>(0)
#define CPPA_LOG_TRACE(unused) static_cast<void>(0)
#define CPPA_LOGF_TRACE(unused) static_cast<void>(0)
#else
#define CPPA_DO_LOG_FUN(level, message) { \
std::ostringstream scoped_oss; scoped_oss << message; \
::cppa::detail::logging::instance()->log( \
level, "NONE", \
__FUNCTION__, __FILE__, __LINE__, scoped_oss.str()); \
} static_cast<void>(0)
#define CPPA_DO_LOG_MEMBER_FUN(level, message) { \
std::ostringstream scoped_oss; scoped_oss << message; \
::cppa::detail::logging::instance()->log( \
level, ::cppa::detail::demangle(typeid(*this)).c_str(), \
__FUNCTION__, __FILE__, __LINE__, scoped_oss.str()); \
} static_cast<void>(0)
#define CPPA_LOG_ERROR(message) CPPA_DO_LOG_MEMBER_FUN("ERROR ", message)
#define CPPA_LOGF_ERROR(message) CPPA_DO_LOG_FUN("ERROR ", message)
#if CPPA_LOG_LEVEL > 0
# define CPPA_LOG_WARNING(message) CPPA_DO_LOG_MEMBER_FUN("WARNING", message)
# define CPPA_LOGF_WARNING(message) CPPA_DO_LOG_FUN("WARNING", message)
#else
# define CPPA_LOG_WARNING(unused) static_cast<void>(0)
# define CPPA_LOGF_WARNING(unused) static_cast<void>(0)
#endif
#if CPPA_LOG_LEVEL > 1
# define CPPA_LOG_INFO(message) CPPA_DO_LOG_MEMBER_FUN("INFO ", message)
# define CPPA_LOGF_INFO(message) CPPA_DO_LOG_FUN("INFO ", message)
#else
# define CPPA_LOG_INFO(unused) static_cast<void>(0)
# define CPPA_LOGF_INFO(unused) static_cast<void>(0)
#endif
#if CPPA_LOG_LEVEL > 2
# define CPPA_LOG_DEBUG(message) CPPA_DO_LOG_MEMBER_FUN("DEBUG ", message)
# define CPPA_LOGF_DEBUG(message) CPPA_DO_LOG_FUN("DEBUG ", message)
#else
# define CPPA_LOG_DEBUG(unused) static_cast<void>(0)
# define CPPA_LOGF_DEBUG(unused) static_cast<void>(0)
#endif
#if CPPA_LOG_LEVEL > 3
# define CPPA_RPAREN )
# define CPPA_LPAREN (
# define CPPA_GET(what) what
# define CPPA_CONCAT_I(lhs,rhs) lhs ## rhs
# define CPPA_CONCAT(lhs,rhs) CPPA_CONCAT_I(lhs,rhs)
# define CPPA_CONCATL(lhs) CPPA_CONCAT(lhs, __LINE__)
# define CPPA_LOG_TRACE(message) \
::std::ostringstream CPPA_CONCATL(cppa_trace_helper_) ; \
CPPA_CONCATL(cppa_trace_helper_) << message ; \
::cppa::detail::logging::trace_helper CPPA_CONCATL(cppa_fun_trace_helper_) \
CPPA_LPAREN ::cppa::detail::demangle \
CPPA_LPAREN typeid CPPA_LPAREN *this CPPA_RPAREN CPPA_RPAREN , \
__func__ , __FILE__ , __LINE__ , \
CPPA_CONCATL(cppa_trace_helper_) .str() CPPA_RPAREN
# define CPPA_LOGF_TRACE(message) \
::std::ostringstream CPPA_CONCATL(cppa_trace_helper_) ; \
CPPA_CONCATL(cppa_trace_helper_) << message ; \
::cppa::detail::logging::trace_helper CPPA_CONCATL(cppa_fun_trace_helper_) \
CPPA_LPAREN "NONE" , \
__func__ , __FILE__ , __LINE__ , \
CPPA_CONCATL(cppa_trace_helper_) .str() CPPA_RPAREN
#else
# define CPPA_LOG_TRACE(unused)
# define CPPA_LOGF_TRACE(unused)
#endif
#define CPPA_LOG_ERROR_IF(stmt,message) \
if (stmt) { CPPA_LOG_ERROR(message); }; static_cast<void>(0)
#define CPPA_LOG_WARNING_IF(stmt,message) \
if (stmt) { CPPA_LOG_WARNING(message); }; static_cast<void>(0)
#define CPPA_LOG_INFO_IF(stmt,message) \
if (stmt) { CPPA_LOG_INFO(message); }; static_cast<void>(0)
#define CPPA_LOG_DEBUG_IF(stmt,message) \
if (stmt) { CPPA_LOG_DEBUG(message); }; static_cast<void>(0)
#define CPPA_LOGF_ERROR_IF(stmt,message) \
if (stmt) { CPPA_LOGF_ERROR(message); }; static_cast<void>(0)
#define CPPA_LOGF_WARNING_IF(stmt,message) \
if (stmt) { CPPA_LOGF_WARNING(message); }; static_cast<void>(0)
#define CPPA_LOGF_INFO_IF(stmt,message) \
if (stmt) { CPPA_LOGF_INFO(message); }; static_cast<void>(0)
#define CPPA_LOGF_DEBUG_IF(stmt,message) \
if (stmt) { CPPA_LOGF_DEBUG(message); }; static_cast<void>(0)
#endif // CPPA_DEBUG
#define CPPA_ARG(arg) #arg << " = " << arg
#define CPPA_TARG(arg, trans) #arg << " = " << trans ( arg )
#define CPPA_MARG(arg, memfun) #arg << " = " << arg . memfun ()
#endif // CPPA_LOGGING_HPP
......@@ -44,6 +44,7 @@ namespace cppa { namespace network { class middleman; } }
namespace cppa { namespace detail {
class logging;
class empty_tuple;
class group_manager;
class abstract_tuple;
......@@ -59,6 +60,8 @@ class singleton_manager {
static void shutdown();
static logging* get_logger();
static scheduler* get_scheduler();
static bool set_scheduler(scheduler*);
......
......@@ -31,10 +31,13 @@
#ifndef CONTINUABLE_READER_HPP
#define CONTINUABLE_READER_HPP
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/network/protocol.hpp"
namespace cppa { namespace network {
class middleman;
......@@ -45,16 +48,16 @@ enum continue_reading_result {
read_continue_later
};
class continuable_reader : public ref_counted {
class continuable_writer;
class continuable_reader : virtual public ref_counted {
public:
/**
* @brief Returns the file descriptor for incoming data.
*/
inline native_socket_type read_handle() const {
return m_read_handle;
}
inline native_socket_type read_handle() const { return m_rd; }
/**
* @brief Reads from {@link read_handle()}.
......@@ -62,36 +65,17 @@ class continuable_reader : public ref_counted {
virtual continue_reading_result continue_reading() = 0;
/**
* @brief Returns @p true if @p this is a {@link peer_acceptor} that
* is assigned to the published actor @p whom.
* @return Casts @p this to a continuable_writer or returns @p nullptr.
*/
virtual bool is_acceptor_of(const actor_ptr& whom) const;
/**
* @brief Returns true if this is a subtype of {@link peer}, i.e.,
* if @p static_cast<peer*>(this) is well-defined.
*/
inline bool is_peer() const {
return m_is_peer;
}
virtual continuable_writer* as_writer();
protected:
continuable_reader(middleman* parent, native_socket_type rd, bool is_peer);
inline middleman* parent() {
return m_parent;
}
inline const middleman* parent() const {
return m_parent;
}
continuable_reader(native_socket_type rd);
private:
bool m_is_peer;
middleman* m_parent;
native_socket_type m_read_handle;
native_socket_type m_rd;
};
......
......@@ -28,42 +28,52 @@
\******************************************************************************/
#ifndef PEER_ACCEPTOR_HPP
#define PEER_ACCEPTOR_HPP
#ifndef CONTINUABLE_WRITER_HPP
#define CONTINUABLE_WRITER_HPP
#include "cppa/network/peer.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace network {
class peer_acceptor : public continuable_reader {
enum continue_writing_result {
write_failure,
write_closed,
write_continue_later,
write_done
};
typedef continuable_reader super;
class continuable_writer : virtual public ref_counted {
public:
typedef ref_counted super;
bool is_acceptor_of(const actor_ptr& whom) const;
public:
protected:
/**
* @brief Returns the file descriptor for outgoing data.
*/
native_socket_type write_handle() const {
return m_wr;
}
peer_acceptor(middleman* parent,
native_socket_type fd,
const actor_ptr& published_actor);
/**
* @brief Writes to {@link write_handle()}.
*/
virtual continue_writing_result continue_writing() = 0;
void add_peer(const peer_ptr& ptr);
protected:
inline const actor_ptr& published_actor() const {
return m_published_actor;
}
continuable_writer(native_socket_type write_handle);
private:
actor_ptr m_published_actor;
native_socket_type m_wr;
};
typedef intrusive_ptr<peer_acceptor> peer_acceptor_ptr;
typedef intrusive_ptr<continuable_writer> continuable_writer_ptr;
} } // namespace cppa::network
#endif // PEER_ACCEPTOR_HPP
#endif // CONTINUABLE_WRITER_HPP
......@@ -28,63 +28,58 @@
\******************************************************************************/
#ifndef IPV4_PEER_HPP
#define IPV4_PEER_HPP
#ifndef CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
#define CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
#include "cppa/process_information.hpp"
#include "cppa/util/buffer.hpp"
#include <map>
#include <cstdint>
#include "cppa/network/peer.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/actor_addressing.hpp"
#include "cppa/process_information.hpp"
namespace cppa { namespace network {
class default_peer_impl : public peer {
class default_protocol;
typedef peer super;
class default_actor_addressing : public actor_addressing {
public:
default_peer_impl(middleman* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr = nullptr);
default_actor_addressing(default_protocol* parent = nullptr);
typedef std::map<actor_id,weak_actor_proxy_ptr> proxy_map;
atom_value technology_id() const;
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);
continue_reading_result continue_reading();
actor_ptr get_or_put(const process_information& parent, actor_id aid);
continue_writing_result continue_writing();
void put(const process_information& parent,
actor_id aid,
const actor_proxy_ptr& proxy);
bool enqueue(const addressed_message& msg);
proxy_map& proxies(process_information& from);
protected:
void erase(process_information& info);
~default_peer_impl();
void erase(process_information& info, actor_id aid);
private:
enum read_state {
// connection just established; waiting for process information
wait_for_process_info,
// wait for the size of the next message
wait_for_msg_size,
// currently reading a message
read_message
};
input_stream_ptr m_in;
output_stream_ptr m_out;
read_state m_state;
process_information_ptr m_peer;
const uniform_type_info* m_meta_msg;
bool m_has_unwritten_data;
util::buffer m_rd_buf;
util::buffer m_wr_buf;
default_protocol* m_parent;
std::map<process_information,proxy_map> m_proxies;
};
} } // namespace cppa::network
#endif // IPV4_PEER_HPP
#endif // CPPA_DEFAULT_ACTOR_ADDRESSING_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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_ACTOR_PROXY_HPP
#define DEFAULT_ACTOR_PROXY_HPP
#include "cppa/actor_proxy.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/detail/abstract_actor.hpp"
namespace cppa { namespace network {
class default_actor_proxy : public detail::abstract_actor<actor_proxy> {
typedef detail::abstract_actor<actor_proxy> super;
public:
default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
const default_protocol_ptr& parent);
void enqueue(actor* sender, any_tuple msg);
void sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
void link_to(const intrusive_ptr<actor>& other);
void unlink_from(const intrusive_ptr<actor>& other);
bool remove_backlink(const intrusive_ptr<actor>& to);
bool establish_backlink(const intrusive_ptr<actor>& to);
void local_link_to(const intrusive_ptr<actor>& other);
void local_unlink_from(const actor_ptr& other);
protected:
~default_actor_proxy();
private:
void forward_msg(const actor_ptr& sender,
any_tuple msg,
message_id_t mid = message_id_t());
default_protocol_ptr m_proto;
};
} } // namespace cppa::network
#endif // DEFAULT_ACTOR_PROXY_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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_DEFAULT_PEER_IMPL_HPP
#define CPPA_DEFAULT_PEER_IMPL_HPP
#include <map>
#include <cstdint>
#include "cppa/actor_proxy.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/weak_intrusive_ptr.hpp"
#include "cppa/process_information.hpp"
#include "cppa/util/buffer.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp"
namespace cppa { namespace network {
class default_protocol;
class default_peer : public continuable_reader, public continuable_writer {
typedef continuable_reader lsuper;
typedef continuable_writer rsuper;
public:
default_peer(default_protocol* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr = nullptr);
continue_reading_result continue_reading();
continue_writing_result continue_writing();
continuable_writer* as_writer();
void enqueue(const addressed_message& msg);
inline bool erase_on_last_proxy_exited() const {
return m_erase_on_last_proxy_exited;
}
inline const process_information& node() const {
return *m_node;
}
protected:
~default_peer();
private:
void disconnected();
enum read_state {
// connection just established; waiting for process information
wait_for_process_info,
// wait for the size of the next message
wait_for_msg_size,
// currently reading a message
read_message
};
default_protocol* m_parent;
input_stream_ptr m_in;
output_stream_ptr m_out;
read_state m_state;
process_information_ptr m_node;
const uniform_type_info* m_meta_msg;
bool m_has_unwritten_data;
util::buffer m_rd_buf;
util::buffer m_wr_buf;
// 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;
partial_function m_content_handler;
void monitor(const actor_ptr& sender, const process_information_ptr& node, actor_id aid);
void kill_proxy(const actor_ptr& sender, const process_information_ptr& node, actor_id aid, std::uint32_t reason);
void link(const actor_ptr& sender, const actor_ptr& ptr);
void unlink(const actor_ptr& sender, const actor_ptr& ptr);
void deliver(const addressed_message& msg);
inline void enqueue(const any_tuple& msg) {
enqueue(addressed_message(nullptr, nullptr, msg));
}
template<typename Arg0, typename Arg1, typename... Args>
inline void enqueue(Arg0&& arg0, Arg1&& arg1, Args&&... args) {
enqueue(make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Arg1>(arg1),
std::forward<Args>(args)...));
}
};
typedef intrusive_ptr<default_peer> default_peer_ptr;
} } // namespace cppa::network
#endif // CPPA_DEFAULT_PEER_IMPL_HPP
......@@ -34,28 +34,36 @@
#include "cppa/actor.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/peer_acceptor.hpp"
#include "cppa/network/continuable_reader.hpp"
namespace cppa { namespace network {
class default_peer_acceptor_impl : public peer_acceptor {
class default_protocol;
typedef peer_acceptor super;
class default_peer_acceptor : public continuable_reader {
typedef continuable_reader super;
public:
continue_reading_result continue_reading();
default_peer_acceptor_impl(middleman* parent,
acceptor_uptr ptr,
const actor_ptr& published_actor);
default_peer_acceptor(default_protocol* parent,
acceptor_uptr ptr,
const actor_ptr& published_actor);
inline const actor_ptr& published_actor() const { return m_pa; }
private:
default_protocol* m_parent;
acceptor_uptr m_ptr;
actor_ptr m_pa;
};
typedef intrusive_ptr<default_peer_acceptor> default_peer_acceptor_ptr;
} } // namespace cppa::detail
#endif // IPV4_PEER_ACCEPTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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/network/protocol.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
namespace cppa { namespace network {
class default_protocol : public protocol {
typedef protocol super;
public:
default_protocol(middleman* parent);
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(io_stream_ptr_pair ioptrs, variant_args args);
inline default_actor_addressing* addressing() { return &m_addressing; }
void register_peer(const process_information& node, default_peer* ptr);
default_peer_ptr get_peer(const process_information& node);
void new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node = nullptr);
void erase_peer(const default_peer_ptr& pptr);
void continue_writer(const default_peer_ptr& pptr);
private:
default_actor_addressing m_addressing;
std::map<actor_ptr,std::vector<default_peer_acceptor_ptr> > m_acceptors;
std::map<process_information,default_peer_ptr> m_peers;
};
typedef intrusive_ptr<default_protocol> default_protocol_ptr;
} } // namespace cppa::network
#endif // DEFAULT_PROTOCOL_HPP
......@@ -34,11 +34,12 @@
#include <map>
#include <vector>
#include <memory>
#include <functional>
#include "cppa/network/peer.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/peer_acceptor.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp"
namespace cppa { namespace detail { class singleton_manager; } }
......@@ -52,7 +53,10 @@ void middleman_loop(middleman_impl*);
class middleman {
// the most popular class in libcppa
friend class peer;
friend class protocol;
friend class peer_acceptor;
friend class singleton_manager;
friend class middleman_overseer;
......@@ -65,28 +69,9 @@ class middleman {
virtual ~middleman();
virtual void publish(std::unique_ptr<acceptor> server,
const actor_ptr& published_actor) = 0;
virtual void add_peer(const io_stream_ptr_pair& io,
const process_information_ptr& node_info) = 0;
virtual void unpublish(const actor_ptr& whom) = 0;
virtual void enqueue(const process_information_ptr& receiving_node,
const addressed_message& message) = 0;
virtual void add_protocol(const protocol_ptr& impl) = 0;
inline void enqueue(const process_information_ptr& receiving_node,
actor_ptr sender,
channel_ptr receiver,
any_tuple msg,
message_id_t id = message_id_t()) {
enqueue(receiving_node,
addressed_message(std::move(sender),
std::move(receiver),
std::move(msg),
id));
}
virtual protocol_ptr protocol(atom_value id) = 0;
protected:
......@@ -95,7 +80,11 @@ class middleman {
virtual void stop() = 0;
virtual void start() = 0;
private:
// to be called from protocol
// runs @p fun in the middleman's event loop
virtual void run_later(std::function<void()> fun) = 0;
// to be called from singleton_manager
......@@ -104,29 +93,33 @@ class middleman {
// to be called from peer
void continue_writing_later(const peer_ptr& ptr);
void register_peer(const process_information& node, const peer_ptr& ptr);
/**
* @pre ptr->as_writer() != nullptr
*/
void continue_writer(const continuable_reader_ptr& ptr);
/**
* @pre ptr->as_writer() != nullptr
*/
void stop_writer(const continuable_reader_ptr& ptr);
// to be called form peer_acceptor or protocol
// to be called form peer_acceptor
void continue_reader(const continuable_reader_ptr& what);
void add(const continuable_reader_ptr& what);
void stop_reader(const continuable_reader_ptr& what);
// to be called from m_handler or middleman_overseer
inline void quit() { m_done = true; }
inline bool done() const { return m_done; }
void erase(const continuable_reader_ptr& what);
continuable_reader_ptr acceptor_of(const actor_ptr& whom);
peer_ptr get_peer(const process_information& node);
// member variables
bool m_done;
std::vector<continuable_reader_ptr> m_readers;
std::map<process_information,peer_ptr> m_peers;
std::unique_ptr<middleman_event_handler> m_handler;
};
......
......@@ -28,78 +28,79 @@
\******************************************************************************/
#ifndef PEER_HPP
#define PEER_HPP
#ifndef CPPA_PROTOCOL_HPP
#define CPPA_PROTOCOL_HPP
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/continuable_reader.hpp"
#include <memory>
#include <functional>
#include <initializer_list>
enum continue_writing_result {
write_failure,
write_closed,
write_continue_later,
write_done
};
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/network/acceptor.hpp"
namespace cppa { namespace network {
class middleman;
class continuable_reader;
class continuable_writer;
/**
* @brief Represents a bidirectional connection to a peer.
*/
class peer : public continuable_reader {
class protocol : public ref_counted {
typedef continuable_reader super;
typedef ref_counted super;
public:
/**
* @brief Returns the file descriptor for outgoing data.
*/
native_socket_type write_handle() const {
return m_write_handle;
}
/**
* @brief Writes to {@link write_handle()}.
*/
virtual continue_writing_result continue_writing() = 0;
/**
* @brief Enqueues @p msg to the list of outgoing messages.
* @returns @p true on success, @p false otherwise.
* @note Implementation should call {@link begin_writing()} and perform IO
* only in its implementation of {@link continue_writing()}.
* @note Returning @p false from this function is interpreted as error
* and causes the middleman to remove this peer.
*/
virtual bool enqueue(const addressed_message& msg) = 0;
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(io_stream_ptr_pair ioptrs,
variant_args args ) = 0;
void run_later(std::function<void()> fun);
protected:
/**
* @brief Tells the middleman to add write_handle() to the list of
* observed sockets and to call continue_writing() if
* write_handle() is ready to write.
* @note Does nothing if write_handle() is already registered for the
* event loop.
*/
void begin_writing();
// note: not thread-safe; call only in run_later functor!
void continue_reader(continuable_reader* what);
// note: not thread-safe; call only in run_later functor!
void continue_writer(continuable_reader* what);
// note: not thread-safe; call only in run_later functor!
void stop_reader(continuable_reader* what);
// note: not thread-safe; call only in run_later functor!
void stop_writer(continuable_reader* what);
void register_peer(const process_information& pinfo);
inline middleman* parent() { return m_parent; }
peer(middleman* parent, native_socket_type rd, native_socket_type wr);
inline const middleman* parent() const { return m_parent; }
private:
native_socket_type m_write_handle;
middleman* m_parent;
};
typedef intrusive_ptr<peer> peer_ptr;
typedef intrusive_ptr<protocol> protocol_ptr;
} } // namespace cppa::network
#endif // PEER_HPP
#endif // CPPA_PROTOCOL_HPP
......@@ -39,7 +39,7 @@
namespace cppa {
// forward declaration
class actor_addressing;
class primitive_variant;
/**
......@@ -53,7 +53,7 @@ class serializer {
public:
serializer() = default;
serializer(actor_addressing* addressing = nullptr);
virtual ~serializer();
......@@ -99,6 +99,12 @@ class serializer {
*/
virtual void write_tuple(size_t num, const primitive_variant* values) = 0;
inline actor_addressing* addressing() { return m_addressing; }
private:
actor_addressing* m_addressing;
};
/**
......
......@@ -44,6 +44,10 @@
#include "cppa/uniform_type_info.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/addressed_message.hpp"
namespace std { class exception; }
namespace cppa {
namespace detail {
......@@ -51,44 +55,42 @@ namespace detail {
std::string to_string_impl(const void* what, const uniform_type_info* utype);
template<typename T>
struct has_cppa_to_string : std::false_type { };
template<>
struct has_cppa_to_string<any_tuple> : std::true_type { };
template<>
struct has_cppa_to_string<network::addressed_message> : std::true_type { };
inline std::string to_string_impl(const T& what) {
return to_string_impl(&what, uniform_typeid<T>());
}
template<>
struct has_cppa_to_string<actor_ptr> : std::true_type { };
} // namespace detail
template<>
struct has_cppa_to_string<group_ptr> : std::true_type { };
inline std::string to_string(const any_tuple& what) {
return detail::to_string_impl(what);
}
template<>
struct has_cppa_to_string<channel_ptr> : std::true_type { };
inline std::string to_string(const network::addressed_message& what) {
return detail::to_string_impl(what);
}
template<>
struct has_cppa_to_string<process_information_ptr> : std::true_type { };
inline std::string to_string(const actor_ptr& what) {
return detail::to_string_impl(what);
}
} // namespace detail
inline std::string to_string(const group_ptr& what) {
return detail::to_string_impl(what);
}
/**
* @brief Converts any of libcppa's core types to a string.
*/
template<typename T>
inline std::string to_string(const T& what,
typename std::enable_if<detail::has_cppa_to_string<T>::value>::type* = 0) {
return detail::to_string_impl(&what, uniform_typeid<T>());
inline std::string to_string(const process_information& what) {
return detail::to_string_impl(what);
}
/**
* @brief Converts an object to a string.
*/
inline std::string to_string(const object& what) {
return detail::to_string_impl(what.value(), what.type());
}
/**
* @brief Converts @p e to a string including the demangled type of e
* and @p e.what().
*/
std::string to_verbose_string(const std::exception& e);
} // namespace cppa
#endif // CPPA_TO_STRING_HPP
......@@ -31,15 +31,18 @@
#ifndef CPPA_WEAK_INTRUSIVE_PTR_HPP
#define CPPA_WEAK_INTRUSIVE_PTR_HPP
#include <cstddef>
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
template<typename T>
class weak_intrusive_ptr {
class weak_intrusive_ptr : util::comparable<weak_intrusive_ptr<T>> {
typedef T::anchor anchor_type;
typedef typename T::weak_ptr_anchor anchor_type;
public:
......@@ -66,6 +69,18 @@ class weak_intrusive_ptr {
return (m_anchor) ? m_anchor->expired() : true;
}
inline ptrdiff_t compare(const weak_intrusive_ptr& other) const {
return m_anchor.compare(other.m_anchor);
}
/**
* @brief Queries whether this weak pointer is invalid, i.e., does not
* point to an instance.
*/
inline bool invalid() const {
return m_anchor == nullptr;
}
private:
intrusive_ptr<anchor_type> m_anchor;
......
......@@ -144,8 +144,8 @@ auto main(int argc, char* argv[]) -> int {
catch (exception& e) {
ostringstream err;
err << "*** exception: group::get(\"" << gid.substr(0, p)
<< "\", \"" << gid.substr(p + 1) << "\") failed, what = "
<< e.what() << endl;
<< "\", \"" << gid.substr(p + 1) << "\") failed; "
<< to_verbose_string(e) << endl;
send(printer, err.str());
}
}
......
......@@ -45,84 +45,7 @@ using namespace std;
namespace cppa {
namespace {
template<typename... Args>
void middleman_enqueue(Args&&... args) {
detail::singleton_manager::get_middleman()->enqueue(forward<Args>(args)...);
}
} // namespace <anonymous>
actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
actor_proxy::actor_proxy(actor_id mid, const process_information_ptr& pptr)
: super(mid, pptr) { }
void actor_proxy::enqueue(actor* sender, any_tuple msg) {
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.get_as<atom_value>(0) == atom("KILL_PROXY")
&& msg.type_at(1) == arr[1]) {
cleanup(msg.get_as<std::uint32_t>(1));
return;
}
middleman_enqueue(parent_process_ptr(), sender, this, std::move(msg));
}
void actor_proxy::sync_enqueue(actor* sender, message_id_t id, any_tuple msg) {
middleman_enqueue(parent_process_ptr(), sender, this, std::move(msg), id);
}
void actor_proxy::link_to(const intrusive_ptr<actor>& other) {
if (link_to_impl(other)) {
// causes remote actor to link to (proxy of) other
middleman_enqueue(parent_process_ptr(),
other,
this,
make_any_tuple(atom("LINK"), other));
}
}
void actor_proxy::local_link_to(const intrusive_ptr<actor>& other) {
link_to_impl(other);
}
void actor_proxy::unlink_from(const intrusive_ptr<actor>& other) {
if (unlink_from_impl(other)) {
// causes remote actor to unlink from (proxy of) other
middleman_enqueue(parent_process_ptr(),
other,
this,
make_any_tuple(atom("UNLINK"), other));
}
}
void actor_proxy::local_unlink_from(const intrusive_ptr<actor>& other) {
unlink_from_impl(other);
}
bool actor_proxy::establish_backlink(const intrusive_ptr<actor>& other) {
bool result = super::establish_backlink(other);
if (result) {
// causes remote actor to unlink from (proxy of) other
middleman_enqueue(parent_process_ptr(),
other,
this,
make_any_tuple(atom("LINK"), other));
}
return result;
}
bool actor_proxy::remove_backlink(const intrusive_ptr<actor>& other) {
bool result = super::remove_backlink(other);
if (result) {
middleman_enqueue(parent_process_ptr(),
nullptr,
this,
make_any_tuple(atom("UNLINK"), actor_ptr(this)));
}
return result;
}
} // namespace cppa
......@@ -38,6 +38,8 @@
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"
#include "cppa/detail/logging.hpp"
namespace {
typedef std::lock_guard<cppa::util::shared_spinlock> exclusive_guard;
......@@ -52,15 +54,19 @@ actor_registry::actor_registry() : m_running(0), m_ids(1) {
}
actor_registry::value_type actor_registry::get_entry(actor_id key) const {
CPPA_LOG_TRACE("key = " << key);
shared_guard guard(m_instances_mtx);
auto i = m_entries.find(key);
if (i != m_entries.end()) {
CPPA_LOG_DEBUG("result = " << i->second.first.get());
return i->second;
}
CPPA_LOG_DEBUG("result = nullptr");
return {nullptr, exit_reason::not_exited};
}
void actor_registry::put(actor_id key, const actor_ptr& value) {
CPPA_LOG_TRACE("key = " << key << ", ptr = " << value.get());
bool add_attachable = false;
if (value != nullptr) {
shared_guard guard(m_instances_mtx);
......@@ -90,6 +96,7 @@ void actor_registry::put(actor_id key, const actor_ptr& value) {
}
void actor_registry::erase(actor_id key, std::uint32_t reason) {
CPPA_LOG_TRACE("key = " << key << ", reason = " << std::hex << reason);
exclusive_guard guard(m_instances_mtx);
auto i = m_entries.find(key);
if (i != m_entries.end()) {
......
......@@ -125,13 +125,13 @@ struct pt_reader {
} // namespace <anonmyous>
binary_deserializer::binary_deserializer(const char* buf, size_t buf_size)
: pos(buf), end(buf + buf_size) {
}
binary_deserializer::binary_deserializer(const char* buf, size_t buf_size,
actor_addressing* addressing)
: super(addressing), pos(buf), end(buf + buf_size) { }
binary_deserializer::binary_deserializer(const char* bbegin, const char* bend)
: pos(bbegin), end(bend) {
}
binary_deserializer::binary_deserializer(const char* bbegin, const char* bend,
actor_addressing* addressing)
: super(addressing), pos(bbegin), end(bend) { }
std::string binary_deserializer::seek_object() {
std::string result;
......
......@@ -110,7 +110,8 @@ class binary_writer {
} // namespace <anonymous>
binary_serializer::binary_serializer(util::buffer* buf) : m_sink(buf) { }
binary_serializer::binary_serializer(util::buffer* buf, actor_addressing* ptr)
: super(ptr), m_sink(buf) { }
void binary_serializer::begin_object(const std::string& tname) {
binary_writer::write_string(m_sink, tname);
......
......@@ -29,16 +29,13 @@
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp"
namespace cppa { namespace network {
continuable_reader::continuable_reader(middleman* parent,
native_socket_type rd,
bool is_peer)
: m_is_peer(is_peer), m_parent(parent), m_read_handle(rd) { }
continuable_reader::continuable_reader(native_socket_type rd) : m_rd(rd) { }
continuable_writer* continuable_reader::as_writer() { return nullptr; }
bool continuable_reader::is_acceptor_of(const actor_ptr&) const {
return false;
}
} } // namespace cppa::network
......@@ -28,20 +28,10 @@
\******************************************************************************/
#include "cppa/network/peer.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/continuable_writer.hpp"
namespace cppa { namespace network {
peer::peer(middleman* parent, native_socket_type rd, native_socket_type wr)
: super(parent, rd, true), m_write_handle(wr) { }
void peer::begin_writing() {
parent()->continue_writing_later(this);
}
void peer::register_peer(const process_information& pinfo) {
parent()->register_peer(pinfo, this);
}
continuable_writer::continuable_writer(native_socket_type wr) : m_wr(wr) { }
} } // namespace cppa::network
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 <cstdint>
#include "cppa/to_string.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/network/default_actor_proxy.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
using namespace std;
namespace cppa { namespace network {
default_actor_addressing::default_actor_addressing(default_protocol* parent)
: m_parent(parent) { }
atom_value default_actor_addressing::technology_id() const {
return atom("DEFAULT");
}
void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
CPPA_LOG_TRACE("sink = " << sink << ", ptr" << ptr.get());
CPPA_REQUIRE(sink != nullptr);
if (ptr == nullptr) {
CPPA_LOG_DEBUG("serialized nullptr");
sink->begin_object("@0");
sink->end_object();
}
else {
// local actor?
if (*ptr->parent_process_ptr() == *process_information::get()) {
detail::singleton_manager::get_actor_registry()->put(ptr->id(), ptr);
}
primitive_variant ptup[2];
ptup[0] = ptr->id();
ptup[1] = ptr->parent_process().process_id();
sink->begin_object("@actor");
sink->write_tuple(2, ptup);
sink->write_raw(process_information::node_id_size,
ptr->parent_process().node_id().data());
sink->end_object();
}
}
actor_ptr default_actor_addressing::read(deserializer* source) {
CPPA_LOG_TRACE("source = " << source);
CPPA_REQUIRE(source != nullptr);
auto cname = source->seek_object();
if (cname == "@0") {
CPPA_LOG_DEBUG("deserialized nullptr");
source->begin_object("@0");
source->end_object();
return nullptr;
}
else if (cname == "@actor") {
primitive_variant ptup[2];
primitive_type ptypes[] = { pt_uint32, pt_uint32 };
process_information::node_id_type nid;
source->begin_object(cname);
source->read_tuple(2, ptypes, ptup);
source->read_raw(process_information::node_id_size, nid.data());
source->end_object();
// local actor?
auto pinf = process_information::get();
if ( pinf->process_id() == cppa::get<uint32_t>(ptup[1])
&& std::equal(pinf->node_id().begin(), pinf->node_id().end(), nid.begin())) {
auto id = cppa::get<uint32_t>(ptup[0]);
return detail::singleton_manager::get_actor_registry()->get(id);
}
else {
process_information tmp(cppa::get<uint32_t>(ptup[1]), nid);
return get_or_put(tmp, cppa::get<uint32_t>(ptup[0]));
}
}
else throw runtime_error("expected type name \"@0\" or \"@actor\"; "
"found: " + cname);
}
size_t default_actor_addressing::count_proxies(const process_information& inf) {
CPPA_LOG_TRACE("inf = " << to_string(inf));
auto i = m_proxies.find(inf);
return (i != m_proxies.end()) ? i->second.size() : 0;
}
actor_ptr default_actor_addressing::get(const process_information& inf,
actor_id aid) {
CPPA_LOG_TRACE("inf = " << to_string(inf) << ", aid = " << aid);
auto& submap = m_proxies[inf];
auto i = submap.find(aid);
if (i != submap.end()) {
auto result = i->second.promote();
CPPA_LOG_INFO_IF(!result, "proxy instance expired");
return result;
}
return nullptr;
}
void default_actor_addressing::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));
auto p = m_parent->get_peer(node);
CPPA_LOG_ERROR_IF(!p, "put a proxy for an unknown peer");
if (p) {
p->enqueue(addressed_message(nullptr, nullptr, make_any_tuple(atom("MONITOR"), process_information::get(), aid)));
}
}
else {
CPPA_LOG_ERROR("a 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) {
actor_proxy_ptr ptr(new default_actor_proxy(aid, new process_information(inf), m_parent));
put(inf, aid, ptr);
result = ptr;
}
return result;
/*
CPPA_LOG_TRACE("inf = " << to_string(inf) << ", aid = " << aid);
if (m_parent == nullptr) return get(inf, aid);
auto& submap = m_proxies[inf];
auto i = submap.find(aid);
if (i == submap.end()) {
actor_proxy_ptr result(new default_actor_proxy(aid, new process_information(inf), m_parent));
CPPA_LOG_DEBUG("created a new proxy instance: " << to_string(result.get()));
submap.insert(make_pair(aid, result));
auto p = m_parent->get_peer(inf);
CPPA_LOG_ERROR_IF(!p, "created a proxy for an unknown peer");
if (p) {
p->enqueue(addressed_message(nullptr, nullptr, make_any_tuple(atom("MONITOR"), process_information::get(), aid)));
}
return result;
}
auto result = i->second.promote();
CPPA_LOG_INFO_IF(!result, "proxy instance expired");
return result;
*/
}
auto default_actor_addressing::proxies(process_information& i) -> proxy_map& {
return m_proxies[i];
}
void default_actor_addressing::erase(process_information& inf) {
CPPA_LOG_TRACE("inf = " << to_string(inf));
m_proxies.erase(inf);
}
void default_actor_addressing::erase(process_information& inf, actor_id aid) {
CPPA_LOG_TRACE("inf = " << to_string(inf) << ", aid = " << aid);
auto i = m_proxies.find(inf);
if (i != m_proxies.end()) {
i->second.erase(aid);
}
}
} } // namespace cppa::network
......@@ -28,101 +28,127 @@
\******************************************************************************/
#include <thread>
#include <cstring>
#include "cppa/atom.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"
#include "cppa/to_string.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/network/default_actor_proxy.hpp"
// thread_specific_ptr
//#include <boost/thread/tss.hpp>
namespace cppa { namespace detail {
#include "cppa/detail/logging.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace { actor_proxy_cache s_proxy_cache; }
using namespace std;
namespace cppa { namespace network {
default_actor_proxy::default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
const default_protocol_ptr& parent)
: super(mid, pinfo), m_proto(parent) { }
default_actor_proxy::~default_actor_proxy() {
CPPA_LOG_TRACE("node = " << to_string(*parent_process_ptr())
<< ", aid = " << id());
auto aid = id();
auto node = parent_process_ptr();
auto proto = m_proto;
proto->run_later([aid, node, proto] {
CPPA_LOGF_TRACE("lambda from ~default_actor_proxy"
<< "; node = " << to_string(*node) << ", aid " << aid
<< ", proto = " << to_string(proto->identifier()));
if (detail::singleton_manager::get_middleman()) {
proto->addressing()->erase(*node, aid);
auto p = proto->get_peer(*node);
if (p && p->erase_on_last_proxy_exited()) {
if (proto->addressing()->count_proxies(*node) == 0) {
proto->erase_peer(p);
}
}
}
// else: middleman already shut down!
});
}
actor_proxy_cache& get_actor_proxy_cache() { return s_proxy_cache; }
void default_actor_proxy::forward_msg(const actor_ptr& sender, any_tuple msg, message_id_t mid) {
CPPA_LOG_TRACE("");
auto node = parent_process_ptr();
actor_ptr receiver = this;
auto proto = m_proto;
m_proto->run_later([proto, node, sender, receiver, msg, mid] {
CPPA_LOGF_TRACE("lambda from default_actor_proxy::forward_msg; "
<< "node = " << to_string(*node)
<< ", sender =" << to_string(sender)
<< ", receiver = " << to_string(receiver)
<< ", proto = " << to_string(proto->identifier()));
auto p = proto->get_peer(*node);
if (p) p->enqueue(addressed_message(sender, receiver, msg, mid));
});
}
actor_proxy_ptr actor_proxy_cache::get_or_put(actor_id aid,
std::uint32_t process_id,
const process_information::node_id_type& node_id) {
key_tuple k{node_id, process_id, aid};
return get_impl(k, true);
void default_actor_proxy::enqueue(actor* sender, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_ARG(sender) << ", " << CPPA_TARG(msg, to_string));
auto& arr = detail::static_types_array<atom_value, uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.get_as<atom_value>(0) == atom("KILL_PROXY")
&& msg.type_at(1) == arr[1]) {
CPPA_LOG_DEBUG("received KILL_PROXY message");
cleanup(msg.get_as<uint32_t>(1));
return;
}
forward_msg(sender, move(msg));
}
actor_proxy_ptr actor_proxy_cache::get(actor_id aid,
std::uint32_t process_id,
const process_information::node_id_type& node_id) {
key_tuple k{node_id, process_id, aid};
return get_impl(k, false);
void default_actor_proxy::sync_enqueue(actor* sender, message_id_t mid, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_ARG(sender) << ", " << CPPA_MARG(mid, integer_value)
<< ", " << CPPA_TARG(msg, to_string));
forward_msg(sender, move(msg), mid);
}
actor_proxy_ptr actor_proxy_cache::get_impl(const key_tuple& key, bool do_put) {
{ // lifetime scope of shared guard
util::shared_lock_guard<util::shared_spinlock> guard{m_lock};
auto i = m_entries.find(key);
if (i != m_entries.end()) {
return i->second;
}
void default_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
// receiving peer will call: this->local_link_to(other)
forward_msg(this, make_any_tuple(atom("LINK"), other));
}
if (!do_put) { return nullptr; }
process_information_ptr peer(new process_information(std::get<1>(key),
std::get<0>(key)));
actor_proxy_ptr result(new actor_proxy(std::get<2>(key), peer));
{ // lifetime scope of exclusive guard
std::lock_guard<util::shared_spinlock> guard{m_lock};
auto i = m_entries.find(key);
if (i != m_entries.end()) {
return i->second;
}
m_entries.insert(std::make_pair(key, result));
}
void default_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
forward_msg(this, make_any_tuple(atom("UNLINK"), other));
}
result->attach_functor([result](std::uint32_t) {
get_actor_proxy_cache().erase(result);
});
auto pself = process_information::get();
singleton_manager::get_middleman()->enqueue(
peer,
nullptr,
nullptr,
make_any_tuple(atom("MONITOR"), pself, std::get<2>(key)));
return result;
}
bool actor_proxy_cache::erase(const actor_proxy_ptr& pptr) {
auto pinfo = pptr->parent_process_ptr();
key_tuple key(pinfo->node_id(), pinfo->process_id(), pptr->id()); {
std::lock_guard<util::shared_spinlock> guard{m_lock};
return m_entries.erase(key) > 0;
bool default_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
forward_msg(this, make_any_tuple(atom("LINK"), other));
return true;
}
return false;
}
bool actor_proxy_cache::key_tuple_less::operator()(const key_tuple& lhs,
const key_tuple& rhs) const {
int cmp_res = strncmp(reinterpret_cast<const char*>(std::get<0>(lhs).data()),
reinterpret_cast<const char*>(std::get<0>(rhs).data()),
process_information::node_id_size);
if (cmp_res < 0) {
bool default_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
forward_msg(this, make_any_tuple(atom("UNLINK"), other));
return true;
}
else if (cmp_res == 0) {
if (std::get<1>(lhs) < std::get<1>(rhs)) {
return true;
}
else if (std::get<1>(lhs) == std::get<1>(rhs)) {
return std::get<2>(lhs) < std::get<2>(rhs);
}
}
return false;
}
void default_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) {
CPPA_LOG_TRACE(CPPA_MARG(other, get));
unlink_from_impl(other);
}
} } // namespace cppa::detail
} } // namespace cppa::network
......@@ -34,59 +34,70 @@
#include "cppa/on.hpp"
#include "cppa/actor.hpp"
#include "cppa/match.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer_impl.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer.hpp"
using namespace std;
namespace cppa { namespace network {
default_peer_impl::default_peer_impl(middleman* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr)
: super(parent, in->read_handle(), out->write_handle()), m_in(in), m_out(out)
default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr)
: lsuper(in->read_handle()), rsuper(out->write_handle())
, m_parent(parent), m_in(in), m_out(out)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_peer(peer_ptr)
, m_node(peer_ptr)
, m_meta_msg(uniform_typeid<addressed_message>())
, m_has_unwritten_data(false) {
m_rd_buf.reset(m_state == wait_for_process_info
? sizeof(uint32_t) + process_information::node_id_size
: 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;
}
default_peer_impl::~default_peer_impl() {
if (m_peer) {
// collect all children (proxies to actors of m_peer)
vector<actor_proxy_ptr> children;
children.reserve(20);
detail::get_actor_proxy_cache().erase_all(m_peer->node_id(),
m_peer->process_id(),
[&](actor_proxy_ptr& pptr) {
children.push_back(move(pptr));
});
default_peer::~default_peer() {
disconnected();
}
void default_peer::disconnected() {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr"));
if (m_node) {
// kill all proxies
for (actor_proxy_ptr& pptr: children) {
pptr->enqueue(nullptr,
make_any_tuple(atom("KILL_PROXY"),
exit_reason::remote_link_unreachable));
auto& children = m_parent->addressing()->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);
}
}
continue_reading_result default_peer_impl::continue_reading() {
continue_reading_result default_peer::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
try { m_rd_buf.append_from(m_in.get()); }
catch (exception&) { return read_failure; }
catch (exception&) {
disconnected();
return read_failure;
}
if (!m_rd_buf.full()) return read_continue_later; // try again later
switch (m_state) {
case wait_for_process_info: {
......@@ -97,21 +108,18 @@ continue_reading_result default_peer_impl::continue_reading() {
memcpy(&process_id, m_rd_buf.data(), sizeof(uint32_t));
memcpy(node_id.data(), m_rd_buf.data() + sizeof(uint32_t),
process_information::node_id_size);
m_peer.reset(new process_information(process_id, node_id));
if (*process_information::get() == *m_peer) {
m_node.reset(new process_information(process_id, node_id));
if (*process_information::get() == *m_node) {
std::cerr << "*** middleman warning: "
"incoming connection from self"
<< std::endl;
return read_failure;
}
register_peer(*m_peer);
CPPA_LOG_DEBUG("read process info: " << to_string(*m_node));
m_parent->register_peer(*m_node, this);
// initialization done
m_state = wait_for_msg_size;
m_rd_buf.reset(sizeof(uint32_t));
//DEBUG("pinfo read: "
// << m_peer->process_id()
// << "@"
// << to_string(m_peer->node_id()));
break;
}
case wait_for_msg_size: {
......@@ -125,107 +133,30 @@ continue_reading_result default_peer_impl::continue_reading() {
case read_message: {
//DEBUG("peer_connection::continue_reading: read_message");
addressed_message msg;
binary_deserializer bd(m_rd_buf.data(), m_rd_buf.size());
binary_deserializer bd(m_rd_buf.data(), m_rd_buf.size(),
m_parent->addressing());
m_meta_msg->deserialize(&msg, &bd);
auto& content = msg.content();
CPPA_LOG_DEBUG("deserialized: " << to_string(msg));
//DEBUG("<-- " << to_string(msg));
match(content) (
// monitor messages are sent automatically whenever
// actor_proxy_cache creates a new proxy
// note: aid is the *original* actor id
on(atom("MONITOR"), arg_match) >> [&](const process_information_ptr& pinfo, actor_id aid) {
if (!pinfo) {
//DEBUG("MONITOR received from invalid peer");
return;
}
auto ar = detail::singleton_manager::get_actor_registry();
auto reg_entry = ar->get_entry(aid);
auto pself = process_information::get();
auto send_kp = [=](uint32_t reason) {
parent()->enqueue(pinfo,
nullptr,
nullptr,
make_any_tuple(
atom("KILL_PROXY"),
pself,
aid,
reason
));
};
if (reg_entry.first == nullptr) {
if (reg_entry.second == exit_reason::not_exited) {
// invalid entry
//DEBUG("MONITOR for an unknown actor received");
}
else {
// this actor already finished execution;
// reply with KILL_PROXY message
send_kp(reg_entry.second);
}
}
else {
reg_entry.first->attach_functor(send_kp);
}
on(atom("MONITOR"), arg_match) >> [&](const process_information_ptr& node, actor_id aid) {
monitor(msg.sender(), node, aid);
},
on(atom("KILL_PROXY"), arg_match) >> [&](const process_information_ptr& peer, actor_id aid, std::uint32_t reason) {
auto& cache = detail::get_actor_proxy_cache();
auto proxy = cache.get(aid,
peer->process_id(),
peer->node_id());
if (proxy) {
proxy->enqueue(nullptr,
make_any_tuple(
atom("KILL_PROXY"), reason));
}
else {
//DEBUG("received KILL_PROXY message but didn't "
// "found matching instance in cache");
}
on(atom("KILL_PROXY"), arg_match) >> [&](const process_information_ptr& node, actor_id aid, std::uint32_t reason) {
kill_proxy(msg.sender(), node, aid, reason);
},
on(atom("LINK"), arg_match) >> [&](const actor_ptr& ptr) {
if (msg.sender()->is_proxy() == false) {
//DEBUG("msg.sender() is not a proxy");
return;
}
auto whom = msg.sender().downcast<actor_proxy>();
if ((whom) && (ptr)) whom->local_link_to(ptr);
link(msg.sender(), ptr);
},
on(atom("UNLINK"), arg_match) >> [](const actor_ptr& ptr) {
if (ptr->is_proxy() == false) {
//DEBUG("msg.sender() is not a proxy");
return;
}
auto whom = ptr.downcast<actor_proxy>();
if ((whom) && (ptr)) whom->local_unlink_from(ptr);
on(atom("UNLINK"), arg_match) >> [&](const actor_ptr& ptr) {
unlink(msg.sender(), ptr);
},
others() >> [&] {
auto receiver = msg.receiver().get();
if (receiver) {
if (msg.id().valid()) {
auto ra = dynamic_cast<actor*>(receiver);
//DEBUG("sync message for actor " << ra->id());
if (ra) {
ra->sync_enqueue(
msg.sender().get(),
msg.id(),
move(msg.content()));
}
else{
//DEBUG("ERROR: sync message to a non-actor");
}
}
else {
//DEBUG("async message (sender is "
// << (msg.sender() ? "valid" : "NULL")
// << ")");
receiver->enqueue(
msg.sender().get(),
move(msg.content()));
}
}
else {
//DEBUG("empty receiver");
}
deliver(msg);
}
);
m_rd_buf.reset(sizeof(uint32_t));
......@@ -240,45 +171,197 @@ continue_reading_result default_peer_impl::continue_reading() {
}
}
continue_writing_result default_peer_impl::continue_writing() {
void default_peer::monitor(const actor_ptr&,
const process_information_ptr& node,
actor_id aid) {
CPPA_LOG_TRACE(CPPA_MARG(node, get) << ", " << CPPA_ARG(aid));
if (!node) {
CPPA_LOG_ERROR("received MONITOR from invalid peer");
return;
}
auto entry = detail::singleton_manager::get_actor_registry()->get_entry(aid);
auto pself = process_information::get();
if (*node == *pself) {
CPPA_LOG_ERROR("received 'MONITOR' from pself");
}
else if (entry.first == nullptr) {
if (entry.second == exit_reason::not_exited) {
CPPA_LOG_ERROR("received MONITOR for unknown "
"actor id: " << aid);
}
else {
CPPA_LOG_DEBUG("received MONITOR for an actor "
"that already finished "
"execution; reply KILL_PROXY");
// this actor already finished execution;
// reply with KILL_PROXY message
// get corresponding peer
enqueue(atom("KILL_PROXY"), pself, aid, entry.second);
}
}
else {
CPPA_LOG_DEBUG("attach functor to " << entry.first.get());
default_protocol_ptr proto = m_parent;
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
CPPA_LOGF_TRACE("lambda from default_peer::monitor");
auto p = proto->get_peer(*node);
if (p) p->enqueue(atom("KILL_PROXY"), pself, aid, reason);
});
});
}
}
void default_peer::kill_proxy(const actor_ptr& sender,
const process_information_ptr& node,
actor_id aid,
std::uint32_t reason) {
CPPA_LOG_TRACE(CPPA_MARG(sender, get)
<< ", " << CPPA_MARG(node, get)
<< ", " << CPPA_ARG(aid)
<< ", " << CPPA_ARG(reason));
if (!node) {
CPPA_LOG_ERROR("node = nullptr");
return;
}
if (sender != nullptr) {
CPPA_LOG_ERROR("sender != nullptr");
return;
}
auto proxy = m_parent->addressing()->get(*node, aid);
if (proxy) {
CPPA_LOG_DEBUG("received KILL_PROXY for " << aid
<< ":" << to_string(*node));
proxy->enqueue(nullptr,
make_any_tuple(
atom("KILL_PROXY"), reason));
}
else {
CPPA_LOG_INFO("received KILL_PROXY message but "
"didn't found a matching instance "
"in proxy cache");
}
}
void default_peer::deliver(const addressed_message& msg) {
CPPA_LOG_TRACE("");
auto receiver = msg.receiver().get();
if (receiver) {
if (msg.id().valid()) {
auto ra = dynamic_cast<actor*>(receiver);
if (ra) {
CPPA_LOG_DEBUG("sync message for actor "
<< ra->id());
ra->sync_enqueue(
msg.sender().get(),
msg.id(),
move(msg.content()));
}
else{
CPPA_LOG_ERROR("sync mesage to non-actor");
}
}
else {
CPPA_LOG_DEBUG("async message with "
<< (msg.sender() ? "" : "in")
<< "valid sender");
receiver->enqueue(
msg.sender().get(),
move(msg.content()));
}
}
else {
CPPA_LOG_ERROR("received message with "
"invalid receiver");
}
}
void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
CPPA_LOG_TRACE(CPPA_MARG(sender, get)
<< ", " << CPPA_MARG(ptr, get));
if (ptr && sender && sender->is_proxy()) {
sender.downcast<actor_proxy>()->local_link_to(ptr);
}
else {
CPPA_LOG_ERROR_IF(!sender,
"received LINK from invalid sender");
CPPA_LOG_ERROR_IF(sender && !sender->is_proxy(),
"sender is not a proxy");
CPPA_LOG_ERROR_IF(!ptr, "received LINK with invalid target");
}
}
void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
if (ptr && sender && sender->is_proxy()) {
sender.downcast<actor_proxy>()->local_unlink_from(ptr);
}
else {
CPPA_LOG_ERROR_IF(!sender,
"received UNLINK from invalid sender");
CPPA_LOG_ERROR_IF(sender && !sender->is_proxy(),
"received UNLINK but sender is not a proxy");
CPPA_LOG_ERROR_IF(!ptr, "received UNLINK with invalid target");
}
}
continue_writing_result default_peer::continue_writing() {
CPPA_LOG_TRACE("");
if (m_has_unwritten_data) {
size_t written;
try { written = m_out->write_some(m_wr_buf.data(), m_wr_buf.size()); }
catch (exception&) { return write_failure; }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
disconnected();
return write_failure;
}
if (written != m_wr_buf.size()) {
CPPA_LOG_DEBUG("tried to write " << m_wr_buf.size() << "bytes, "
<< "only " << written << " bytes written");
m_wr_buf.erase_leading(written);
return write_continue_later;
}
else {
m_wr_buf.reset();
m_has_unwritten_data = false;
CPPA_LOG_DEBUG("write done, " << written << "bytes written");
return write_done;
}
}
CPPA_LOG_DEBUG("nothing to write (done)");
return write_done;
}
bool default_peer_impl::enqueue(const addressed_message& msg) {
binary_serializer bs(&m_wr_buf);
continuable_writer* default_peer::as_writer() {
return this;
}
void default_peer::enqueue(const addressed_message& msg) {
CPPA_LOG_TRACE("");
binary_serializer bs(&m_wr_buf, m_parent->addressing());
uint32_t size = 0;
auto before = m_wr_buf.size();
m_wr_buf.write(sizeof(uint32_t), &size, util::grow_if_needed);
try { bs << msg; }
catch (exception& e) {
cerr << "*** exception in default_peer_impl::enqueue; "
<< detail::demangle(typeid(e)) << ", what(): " << e.what()
CPPA_LOG_ERROR(to_verbose_string(e));
cerr << "*** exception in default_peer::enqueue; "
<< to_verbose_string(e)
<< endl;
return false;
return;
}
CPPA_LOG_DEBUG("serialized: " << to_string(msg));
size = (m_wr_buf.size() - before) - sizeof(std::uint32_t);
// update size in buffer
memcpy(m_wr_buf.data() + before, &size, sizeof(std::uint32_t));
CPPA_LOG_DEBUG_IF(m_has_unwritten_data, "still registered for writing");
if (!m_has_unwritten_data) {
CPPA_LOG_DEBUG("register for writing");
m_has_unwritten_data = true;
begin_writing();
m_parent->continue_writer(this);
}
return true;
}
} } // namespace cppa::network
......@@ -31,27 +31,35 @@
#include <iostream>
#include <exception>
#include "cppa/to_string.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/default_peer_impl.hpp"
#include "cppa/network/default_peer_acceptor_impl.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/demangle.hpp"
using namespace std;
namespace cppa { namespace network {
default_peer_acceptor_impl::default_peer_acceptor_impl(middleman* mm,
default_peer_acceptor::default_peer_acceptor(default_protocol* parent,
acceptor_uptr aur,
const actor_ptr& pa)
: super(mm, aur->file_handle(), pa), m_ptr(std::move(aur)) { }
: super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur)), m_pa(pa) { }
continue_reading_result default_peer_acceptor_impl::continue_reading() {
continue_reading_result default_peer_acceptor::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
option<io_stream_ptr_pair> opt;
try { opt = m_ptr->try_accept_connection(); }
catch (...) { return read_failure; }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
return read_failure;
}
if (opt) {
auto& pair = *opt;
auto& pself = process_information::get();
......@@ -62,12 +70,12 @@ continue_reading_result default_peer_acceptor_impl::continue_reading() {
pair.second->write(&process_id, sizeof(uint32_t));
pair.second->write(pself->node_id().data(),
pself->node_id().size());
add_peer(new default_peer_impl(parent(), pair.first, pair.second));
m_parent->new_peer(pair.first, pair.second);
}
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
cerr << "*** exception while sending actor and process id; "
<< detail::demangle(typeid(e))
<< ", what(): " << e.what()
<< to_verbose_string(e)
<< endl;
}
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* 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 3 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 <future>
#include <cstdint>
#include <iostream>
#include "cppa/to_string.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
using namespace std;
using namespace cppa::detail;
namespace cppa { namespace network {
default_protocol::default_protocol(middleman* parent)
: super(parent), m_addressing(this) { }
atom_value default_protocol::identifier() const {
return atom("DEFAULT");
}
void default_protocol::publish(const actor_ptr& whom, variant_args args) {
CPPA_LOG_TRACE("whom: " << to_string(whom)
<< ", args.size() = " << args.size());
if (!whom) return;
CPPA_REQUIRE(args.size() == 2);
auto i = args.begin();
auto port = get<uint16_t>(*i++);
auto& addr = get<string>(*i);
publish(whom,
unique_ptr<acceptor>(ipv4_acceptor::create(port, addr.c_str())),
{});
}
void default_protocol::publish(const actor_ptr& whom,
std::unique_ptr<acceptor> ptr,
variant_args args ) {
CPPA_LOG_TRACE("whom: " << to_string(whom)
<< ", ptr = " << ptr.get()
<< ", args.size() = " << args.size());
if (!whom) return;
CPPA_REQUIRE(args.size() == 0);
static_cast<void>(args); // keep compiler happy
singleton_manager::get_actor_registry()->put(whom->id(), whom);
default_protocol_ptr proto = this;
default_peer_acceptor_ptr impl(new default_peer_acceptor(this,
move(ptr),
whom));
run_later([=] {
CPPA_LOGF_TRACE("lambda from default_protocol::publish");
proto->continue_reader(impl.get());
proto->m_acceptors[whom].push_back(impl);
});
}
void default_protocol::unpublish(const actor_ptr& whom) {
CPPA_LOG_TRACE("whom = " << to_string(whom));
default_protocol_ptr proto = this;
run_later([=] {
CPPA_LOGF_TRACE("lambda from default_protocol::unpublish");
auto& acceptors = m_acceptors[whom];
for (auto& ptr : acceptors) {
proto->stop_reader(ptr.get());
}
m_acceptors.erase(whom);
});
}
void default_protocol::register_peer(const process_information& node,
default_peer* ptr) {
CPPA_LOG_TRACE("node = " << to_string(node) << ", ptr = " << ptr);
auto& ptrref = m_peers[node];
if (ptrref) {
CPPA_LOG_ERROR("peer already defined");
cerr << "*** peer already defined!" << endl;
}
else ptrref = ptr;
}
default_peer_ptr default_protocol::get_peer(const process_information& n) {
CPPA_LOG_TRACE("n = " << to_string(n));
auto e = end(m_peers);
auto i = m_peers.find(n);
if (i != e) {
CPPA_LOG_DEBUG("result = " << i->second.get());
return i->second;
}
CPPA_LOG_DEBUG("result = nullptr");
return nullptr;
}
actor_ptr default_protocol::remote_actor(variant_args args) {
CPPA_LOG_TRACE("args.size() = " << args.size());
CPPA_REQUIRE(args.size() == 2);
auto i = args.begin();
auto port = get<uint16_t>(*i++);
auto& host = get<string>(*i);
auto io = ipv4_io_stream::connect_to(host.c_str(), port);
return remote_actor(io_stream_ptr_pair(io, io), {});
}
struct remote_actor_result { remote_actor_result* next; actor_ptr value; };
actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
variant_args args ) {
CPPA_LOG_TRACE("io = {" << io.first.get() << ", " << io.second.get() << "}, "
<< " << args.size() = " << args.size());
CPPA_REQUIRE(args.size() == 0);
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());
std::uint32_t remote_aid;
std::uint32_t peer_pid;
process_information::node_id_type peer_node_id;
io.first->read(&remote_aid, sizeof(remote_aid));
io.first->read(&peer_pid, sizeof(std::uint32_t));
io.first->read(peer_node_id.data(), peer_node_id.size());
process_information_ptr pinfptr(new process_information(peer_pid, peer_node_id));
if (*pinf == *pinfptr) {
// dude, this is not a remote actor, it's a local actor!
CPPA_LOG_ERROR("remote_actor() called to access a local actor");
# ifndef CPPA_DEBUG
std::cerr << "*** warning: remote_actor() called to access a local actor\n"
<< std::flush;
# endif
return singleton_manager::get_actor_registry()->get(remote_aid);
}
default_protocol_ptr proto = this;
intrusive::single_reader_queue<remote_actor_result> q;
run_later([proto, io, pinfptr, remote_aid, &q] {
CPPA_LOGF_TRACE("lambda from default_protocol::remote_actor");
auto pp = proto->get_peer(*pinfptr);
CPPA_LOGF_INFO_IF(pp, "connection already exists (re-use old one)");
if (!pp) proto->new_peer(io.first, io.second, pinfptr);
auto res = proto->addressing()->get_or_put(*pinfptr, remote_aid);
q.push_back(new remote_actor_result{0, res});
});
unique_ptr<remote_actor_result> result(q.pop());
CPPA_LOGF_DEBUG("result = " << result->value.get());
return result->value;
}
void default_protocol::erase_peer(const default_peer_ptr& pptr) {
CPPA_REQUIRE(pptr != nullptr);
CPPA_LOG_TRACE("pptr = " << pptr.get()
<< ", pptr->node() = " << to_string(pptr->node()));
stop_reader(pptr.get());
auto i = m_peers.find(pptr->node());
if (i != m_peers.end()) {
CPPA_LOG_DEBUG_IF(i->second != pptr, "node " << to_string(pptr->node())
<< " does not exist in m_peers");
if (i->second == pptr) {
m_peers.erase(i);
}
}
}
void default_protocol::new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node) {
CPPA_LOG_TRACE("");
default_peer_ptr ptr(new default_peer(this, in, out, node));
continue_reader(ptr.get());
if (node) register_peer(*node, ptr.get());
}
void default_protocol::continue_writer(const default_peer_ptr& pptr) {
CPPA_LOG_TRACE(CPPA_MARG(pptr, get));
super::continue_writer(pptr.get());
}
} } // namespace cppa::network
......@@ -37,8 +37,9 @@
namespace cppa {
deserializer::~deserializer() {
}
deserializer::deserializer(actor_addressing* aa) : m_addressing(aa) { }
deserializer::~deserializer() { }
deserializer& operator>>(deserializer& d, object& what) {
std::string tname = d.peek_object();
......
......@@ -28,81 +28,113 @@
\******************************************************************************/
#ifndef CPPA_ACTOR_PROXY_CACHE_HPP
#define CPPA_ACTOR_PROXY_CACHE_HPP
#include <mutex>
#include <ctime>
#include <thread>
#include <string>
#include <limits>
#include <vector>
#include <functional>
#include <fstream>
#include <algorithm>
#ifndef CPPA_WINDOWS
#include <unistd.h>
#include <sys/types.h>
#endif
#include "cppa/actor_proxy.hpp"
#include "cppa/process_information.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/util/shared_spinlock.hpp"
using namespace std;
namespace cppa { namespace detail {
class actor_proxy_cache {
namespace {
template<size_t RawSize>
void replace_all(string& str, const char (&before)[RawSize], const char* after) {
// end(before) - 1 points to the null-terminator
auto i = search(begin(str), end(str), begin(before), end(before) - 1);
while (i != end(str)) {
str.replace(i, i + RawSize - 1, after);
i = search(begin(str), end(str), begin(before), end(before) - 1);
}
}
struct log_event {
log_event* next;
string msg;
};
class logging_impl : public logging {
public:
// returns existing instance if available or crates a new one
actor_proxy_ptr get_or_put(actor_id aid,
std::uint32_t process_id,
const process_information::node_id_type& node_id);
actor_proxy_ptr get(actor_id aid,
std::uint32_t process_id,
const process_information::node_id_type& node_id);
// @returns true if pptr was successfully removed, false otherwise
bool erase(const actor_proxy_ptr& pptr);
template<typename Fun>
void erase_all(const process_information::node_id_type& nid,
std::uint32_t process_id,
Fun fun) {
key_tuple lb{nid, process_id, std::numeric_limits<actor_id>::min()};
key_tuple ub{nid, process_id, std::numeric_limits<actor_id>::max()};
{ // lifetime scope of guard
std::lock_guard<util::shared_spinlock> guard(m_lock);
auto e = m_entries.end();
auto first = m_entries.lower_bound(lb);
if (first != e) {
auto last = m_entries.upper_bound(ub);
for (auto i = first; i != last; ++i) {
fun(i->second);
}
m_entries.erase(first, last);
void start() {
m_thread = thread([this] { (*this)(); });
}
void stop() {
log("DEBUG", "logging", "stop", __FILE__, __LINE__, "shutting down");
// an empty string means: shut down
m_queue.push_back(new log_event{0, ""});
m_thread.join();
}
void operator()() {
ostringstream fname;
fname << "libcppa_" << getpid() << "_" << time(0) << ".log";
fstream out(fname.str().c_str(), ios::out);
unique_ptr<log_event> event;
for (;;) {
event.reset(m_queue.pop());
if (event->msg.empty()) {
out.close();
return;
}
else out << event->msg << flush;
}
}
private:
void log(const char* level,
const char* c_class_name,
const char* function_name,
const char* c_full_file_name,
int line_num,
const std::string &msg) {
string class_name = c_class_name;
replace_all(class_name, "::", ".");
string file_name;
string full_file_name = c_full_file_name;
auto i = find(full_file_name.rbegin(), full_file_name.rend(), '/').base();
if (i == full_file_name.end()) {
file_name = move(full_file_name);
}
else {
file_name = string(i, full_file_name.end());
}
ostringstream line;
line << time(0) << " "
<< level << " "
<< this_thread::get_id() << " "
<< class_name << " "
<< function_name << " "
<< file_name << ":" << line_num << " "
<< msg
<< endl;
m_queue.push_back(new log_event{0, line.str()});
}
typedef std::tuple<process_information::node_id_type, // node id
std::uint32_t, // process id
actor_id> // (remote) actor id
key_tuple;
private:
struct key_tuple_less {
bool operator()(const key_tuple& lhs, const key_tuple& rhs) const;
};
thread m_thread;
intrusive::single_reader_queue<log_event> m_queue;
util::shared_spinlock m_lock;
std::map<key_tuple, actor_proxy_ptr, key_tuple_less> m_entries;
};
actor_proxy_ptr get_impl(const key_tuple& key, bool do_put);
} // namespace <anonymous>
logging::~logging() { }
};
logging* logging::instance() { return singleton_manager::get_logger(); }
// get the thread-local cache object
actor_proxy_cache& get_actor_proxy_cache();
logging* logging::create_singleton() { return new logging_impl; }
} } // namespace cppa::detail
#endif // CPPA_ACTOR_PROXY_CACHE_HPP
......@@ -34,6 +34,7 @@
#include <cstring>
#include <sstream>
#include <iostream>
#include <stdexcept>
#include "cppa/on.hpp"
#include "cppa/actor.hpp"
......@@ -53,27 +54,14 @@
#include "cppa/network/middleman.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/default_peer_impl.hpp"
#include "cppa/network/default_peer_acceptor_impl.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
//#define VERBOSE_MIDDLEMAN
#ifdef VERBOSE_MIDDLEMAN
#define DEBUG(arg) { \
ostringstream oss; \
oss << "[process id: " \
<< cppa::process_information::get()->process_id() \
<< "] " << arg << endl; \
cout << oss.str() << flush; \
} (void) 0
#else
#define DEBUG(unused) ((void) 0)
#endif
#include "cppa/intrusive/single_reader_queue.hpp"
// use epoll event-loop implementation on Linux if not explicitly overriden
// by using CPPA_POLL_IMPL, use (default) poll implementation otherwise
......@@ -95,16 +83,6 @@ namespace {
const size_t ui32_size = sizeof(uint32_t);
template<typename T, typename... Args>
void call_ctor(T& var, Args&&... args) {
new (&var) T (forward<Args>(args)...);
}
template<typename T>
void call_dtor(T& var) {
var.~T();
}
template<class Container, class Element>
void erase_from(Container& haystack, const Element& needle) {
typedef typename Container::value_type value_type;
......@@ -122,73 +100,27 @@ void erase_from_if(Container& container, const UnaryPredicate& predicate) {
if (i != last) container.erase(i);
}
} // namespace <anonmyous>
enum class middleman_message_type {
add_peer,
publish,
unpublish,
outgoing_message,
shutdown
};
struct middleman_message {
middleman_message* next;
const middleman_message_type type;
union {
std::pair<io_stream_ptr_pair,process_information_ptr> new_peer;
std::pair<std::unique_ptr<acceptor>, actor_ptr> new_published_actor;
actor_ptr published_actor;
std::pair<process_information_ptr,addressed_message> out_msg;
};
class middleman_event {
middleman_message() : next(0), type(middleman_message_type::shutdown) { }
friend class intrusive::single_reader_queue<middleman_event>;
middleman_message(io_stream_ptr_pair a0, process_information_ptr a1)
: next(0), type(middleman_message_type::add_peer) {
call_ctor(new_peer, move(a0), move(a1));
}
public:
middleman_message(unique_ptr<acceptor> a0, actor_ptr a1)
: next(0), type(middleman_message_type::publish) {
call_ctor(new_published_actor, move(a0), move(a1));
}
template<typename Arg>
middleman_event(Arg&& arg) : next(0), fun(forward<Arg>(arg)) { }
middleman_message(actor_ptr a0)
: next(0), type(middleman_message_type::unpublish) {
call_ctor(published_actor, move(a0));
}
inline void operator()() { fun(); }
middleman_message(process_information_ptr a0, addressed_message a1)
: next(0), type(middleman_message_type::outgoing_message) {
call_ctor(out_msg, move(a0), move(a1));
}
private:
~middleman_message() {
switch (type) {
case middleman_message_type::add_peer:
call_dtor(new_peer);
break;
case middleman_message_type::publish:
call_dtor(new_published_actor);
break;
case middleman_message_type::unpublish:
call_dtor(published_actor);
break;
case middleman_message_type::outgoing_message:
call_dtor(out_msg);
break;
default: break;
}
}
middleman_event* next;
function<void()> fun;
template<typename... Args>
static inline std::unique_ptr<middleman_message> create(Args&&... args) {
return std::unique_ptr<middleman_message>(new middleman_message(std::forward<Args>(args)...));
}
};
typedef intrusive::single_reader_queue<middleman_message> middleman_queue;
typedef intrusive::single_reader_queue<middleman_event> middleman_queue;
} // namespace <anonymous>
class middleman_impl : public middleman {
......@@ -196,24 +128,35 @@ class middleman_impl : public middleman {
public:
middleman_impl() { }
void publish(std::unique_ptr<acceptor> server, const actor_ptr& aptr) {
enqueue_message(middleman_message::create(move(server), aptr));
middleman_impl() {
m_protocols.insert(make_pair(atom("DEFAULT"),
new network::default_protocol(this)));
}
void add_peer(const io_stream_ptr_pair& io,
const process_information_ptr& node_info) {
enqueue_message(middleman_message::create(io, node_info));
void add_protocol(const protocol_ptr& impl) {
if (!impl) {
CPPA_LOG_ERROR("impl == nullptr");
throw std::invalid_argument("impl == nullptr");
}
CPPA_LOG_TRACE("identifier = " << to_string(impl->identifier()));
std::lock_guard<util::shared_spinlock> guard(m_protocols_lock);
m_protocols.insert(make_pair(impl->identifier(), impl));
}
void unpublish(const actor_ptr& whom) {
enqueue_message(middleman_message::create(whom));
protocol_ptr protocol(atom_value id) {
util::shared_lock_guard<util::shared_spinlock> guard(m_protocols_lock);
auto i = m_protocols.find(id);
return (i != m_protocols.end()) ? i->second : nullptr;
}
void enqueue(const process_information_ptr& node,
const addressed_message& msg) {
enqueue_message(middleman_message::create(node, msg));
void run_later(function<void()> fun) {
CPPA_LOG_TRACE("");
m_queue._push_back(new middleman_event(move(fun)));
atomic_thread_fence(memory_order_seq_cst);
uint8_t dummy = 0;
if (write(m_pipe_write, &dummy, sizeof(dummy)) != sizeof(dummy)) {
CPPA_CRITICAL("cannot write to pipe");
}
}
protected:
......@@ -225,11 +168,12 @@ class middleman_impl : public middleman {
m_pipe_write = pipefds[1];
detail::fd_util::nonblocking(m_pipe_read, true);
// start threads
m_thread = std::thread([this] { middleman_loop(this); });
m_thread = thread([this] { middleman_loop(this); });
}
void stop() {
enqueue_message(middleman_message::create());
run_later([this] { this->m_done = true; });
//enqueue_message(middleman_message::create());
m_thread.join();
close(m_pipe_read);
close(m_pipe_write);
......@@ -237,19 +181,13 @@ class middleman_impl : public middleman {
private:
std::thread m_thread;
thread m_thread;
native_socket_type m_pipe_read;
native_socket_type m_pipe_write;
middleman_queue m_queue;
void enqueue_message(std::unique_ptr<middleman_message> msg) {
m_queue._push_back(msg.release());
std::atomic_thread_fence(std::memory_order_seq_cst);
std::uint8_t dummy = 0;
if (write(m_pipe_write, &dummy, sizeof(dummy)) != sizeof(dummy)) {
CPPA_CRITICAL("cannot write to pipe");
}
}
util::shared_spinlock m_protocols_lock;
map<atom_value,protocol_ptr> m_protocols;
};
......@@ -263,12 +201,11 @@ class middleman_overseer : public continuable_reader {
public:
middleman_overseer(middleman* parent, int pipe_fd, middleman_queue& q)
: super(parent, pipe_fd, false), m_queue(q) { }
middleman_overseer(int pipe_fd, middleman_queue& q)
: super(pipe_fd), m_queue(q) { }
continue_reading_result continue_reading() {
//DEBUG("middleman_overseer::continue_reading");
static constexpr size_t num_dummies = 256;
static constexpr size_t num_dummies = 64;
uint8_t dummies[num_dummies];
auto read_result = ::read(read_handle(), dummies, num_dummies);
if (read_result < 0) {
......@@ -277,61 +214,18 @@ class middleman_overseer : public continuable_reader {
return read_continue_later;
}
else {
CPPA_LOG_ERROR("cannot read from pipe");
CPPA_CRITICAL("cannot read from pipe");
}
}
atomic_thread_fence(memory_order_seq_cst);
for (int i = 0; i < read_result; ++i) {
unique_ptr<middleman_message> msg(m_queue.try_pop());
if (!msg) { CPPA_CRITICAL("nullptr dequeued"); }
switch (msg->type) {
case middleman_message_type::add_peer: {
DEBUG("middleman_overseer: add_peer: "
<< to_string(*(msg->new_peer.second)));
auto& new_peer = msg->new_peer;
auto& io_ptrs = new_peer.first;
peer_ptr p(new default_peer_impl(parent(),
io_ptrs.first,
io_ptrs.second,
new_peer.second));
parent()->add(p);
parent()->register_peer(*new_peer.second, p);
break;
}
case middleman_message_type::publish: {
DEBUG("middleman_overseer: publish");
auto& ptrs = msg->new_published_actor;
parent()->add(new default_peer_acceptor_impl(parent(),
move(ptrs.first),
ptrs.second));
break;
}
case middleman_message_type::unpublish: {
if (msg->published_actor) {
//DEBUG("middleman_overseer: unpublish actor id "
// << msg->published_actor->id());
auto channel = parent()->acceptor_of(msg->published_actor);
if (channel) parent()->erase(channel);
}
break;
}
case middleman_message_type::outgoing_message: {
//DEBUG("middleman_overseer: outgoing_message");
auto& target_peer = msg->out_msg.first;
auto& out_msg = msg->out_msg.second;
CPPA_REQUIRE(target_peer != nullptr);
auto p = parent()->get_peer(*target_peer);
if (!p) { DEBUG("message to an unknown peer: "
<< to_string(out_msg)); }
else p->enqueue(out_msg);
break;
}
case middleman_message_type::shutdown: {
DEBUG("middleman: shutdown");
parent()->quit();
break;
}
unique_ptr<middleman_event> msg(m_queue.try_pop());
if (!msg) {
CPPA_LOG_ERROR("nullptr dequeued");
CPPA_CRITICAL("nullptr dequeued");
}
(*msg)();
}
return read_continue_later;
}
......@@ -354,54 +248,166 @@ static constexpr event_bitmask error = 0x04;
} // namespace event
typedef std::tuple<native_socket_type,continuable_reader_ptr,event_bitmask>
fd_meta_info;
inline const char* eb2str(event_bitmask e) {
switch (e) {
default: return "INVALID";
case event::none: return "event::none";
case event::read: return "event::read";
case event::write: return "event::write";
case event::both: return "event::both";
case event::error: return "event::error";
}
}
struct fd_meta_info {
native_socket_type fd;
continuable_reader_ptr ptr;
event_bitmask mask;
fd_meta_info(native_socket_type a0,
const continuable_reader_ptr& a1,
event_bitmask a2)
: fd(a0), ptr(a1), mask(a2) { }
};
enum class fd_meta_event { add, erase, mod };
struct fd_less {
inline bool operator()(const fd_meta_info& lhs, native_socket_type rhs) const {
return lhs.fd < rhs;
}
};
class middleman_event_handler_base {
public:
typedef vector<fd_meta_info> vector_type;
virtual ~middleman_event_handler_base() { }
void add_later(const continuable_reader_ptr& ptr, event_bitmask e) {
append(m_additions, ptr, e);
void alteration(const continuable_reader_ptr& ptr, event_bitmask e, fd_meta_event etype) {
native_socket_type fd;
switch (e) {
case event::read:
fd = ptr->read_handle();
break;
case event::write: {
auto wptr = ptr->as_writer();
if (wptr) fd = wptr->write_handle();
else {
CPPA_LOG_ERROR("ptr->as_writer() returned nullptr");
return;
}
break;
}
case event::both: {
fd = ptr->read_handle();
auto wptr = ptr->as_writer();
if (wptr) {
auto wrfd = wptr->write_handle();
if (fd != wrfd) {
CPPA_LOG_DEBUG("read_handle != write_handle, split "
"into two function calls");
// split into two function calls
e = event::read;
alteration(ptr, event::write, etype);
}
}
else {
CPPA_LOG_ERROR("ptr->as_writer() returned nullptr");
return;
}
break;
}
default:
CPPA_LOG_ERROR("invalid bitmask");
return;
}
/*
auto last = end(m_meta);
auto iter = find_meta(fd);
CPPA_LOG_ERROR_IF(iter == last && e == event::none,
"nothing to delete (no match)");
event_bitmask old = (iter != last) ? iter->mask : event::none;
auto next = cm(old, e);
if (next != old) {
m_alterations.emplace_back(fd, ptr, next);
}
*/
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void add(const continuable_reader_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void erase(const continuable_reader_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
void erase_later(const continuable_reader_ptr& ptr, event_bitmask e) {
append(m_subtractions, ptr, e);
inline event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) {
CPPA_REQUIRE(op == fd_meta_event::add || op == fd_meta_event::erase);
return (op == fd_meta_event::add) ? old | arg : old & ~arg;
}
void update() {
CPPA_LOG_TRACE("");
for (auto& elem_pair : m_alterations) {
auto& elem = elem_pair.first;
auto old = event::none;
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, elem.fd, m_less);
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
CPPA_LOG_DEBUG("new bitmask for "
<< elem.ptr.get() << ": " << eb2str(mask));
if (iter == last || iter->fd != elem.fd) {
CPPA_LOG_INFO_IF(mask == event::none,
"cannot erase " << ptr
<< " (not found in m_meta)");
if (mask != event::none) {
m_meta.insert(iter, elem);
handle_event(fd_meta_event::add, elem.fd,
event::none, mask, ptr);
}
}
else if (iter->fd == elem.fd) {
CPPA_REQUIRE(iter->ptr == elem.ptr);
if (mask == event::none) {
m_meta.erase(iter);
handle_event(fd_meta_event::erase, elem.fd, old, mask, ptr);
}
else {
iter->mask = mask;
handle_event(fd_meta_event::mod, elem.fd, old, mask, ptr);
}
}
}
m_alterations.clear();
}
protected:
vector<fd_meta_info> m_additions;
vector<fd_meta_info> m_subtractions;
virtual void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_reader* ptr) = 0;
fd_less m_less;
vector_type m_meta; // this vector is *always* sorted
vector<pair<fd_meta_info,fd_meta_event> > m_alterations;
private:
void append(vector<fd_meta_info>& vec, const continuable_reader_ptr& ptr, event_bitmask e) {
CPPA_REQUIRE(ptr != nullptr);
CPPA_REQUIRE(e == event::read || e == event::write || e == event::both);
if (e == event::read || (e == event::both && !ptr->is_peer())) {
// ignore event::write unless ptr->is_peer
vec.emplace_back(ptr->read_handle(), ptr, event::read);
}
else if (e == event::write) {
CPPA_REQUIRE(ptr->is_peer());
auto dptr = static_cast<peer*>(ptr.get());
vec.emplace_back(dptr->write_handle(), ptr, event::write);
}
else { // e == event::both && ptr->is_peer()
CPPA_REQUIRE(ptr->is_peer());
CPPA_REQUIRE(e == event::both);
auto dptr = static_cast<peer*>(ptr.get());
auto rd = dptr->read_handle();
auto wr = dptr->write_handle();
if (rd == wr) vec.emplace_back(wr, ptr, event::both);
else {
vec.emplace_back(wr, ptr, event::write);
vec.emplace_back(rd, ptr, event::read);
}
}
vector_type::iterator find_meta(native_socket_type fd) {
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, fd, m_less);
return (iter != last && iter->fd == fd) ? iter : last;
}
};
......@@ -431,7 +437,7 @@ class event_iterator_impl {
}
inline continue_writing_result continue_writing() {
return static_cast<peer*>(ptr())->continue_writing();
return ptr()->as_writer()->continue_writing();
}
inline bool equal_to(const event_iterator_impl& other) const {
......@@ -463,10 +469,8 @@ inline bool operator!=(const event_iterator_impl<Iter,Access>& lhs,
#ifdef CPPA_POLL_IMPL
typedef vector<pollfd> pollfd_vector;
typedef vector<fd_meta_info> fd_meta_vector;
typedef pair<pollfd_vector::iterator,fd_meta_vector::iterator> pfd_iterator;
typedef pair<vector<pollfd>::iterator,vector<fd_meta_info>::iterator>
pfd_iterator;
#ifndef POLLRDHUP
#define POLLRDHUP POLLHUP
......@@ -474,10 +478,7 @@ typedef pair<pollfd_vector::iterator,fd_meta_vector::iterator> pfd_iterator;
struct pfd_access {
inline void advance(pfd_iterator& i) const {
++(i.first);
++(i.second);
}
inline void advance(pfd_iterator& i) const { ++(i.first); ++(i.second); }
inline event_bitmask type(const pfd_iterator& i) const {
auto revents = i.first->revents;
......@@ -493,7 +494,7 @@ struct pfd_access {
}
inline continuable_reader* ptr(pfd_iterator& i) const {
return std::get<1>(*(i.second)).get();
return i.second->ptr.get();
}
inline bool equal(const pfd_iterator& lhs, const pfd_iterator& rhs) const {
......@@ -506,6 +507,21 @@ struct pfd_access {
typedef event_iterator_impl<pfd_iterator,pfd_access> event_iterator;
struct pollfd_less {
inline bool operator()(const pollfd& lhs, native_socket_type rhs) const {
return lhs.fd < rhs;
}
};
short to_poll_bitmask(event_bitmask mask) {
switch (mask) {
case event::read: return POLLIN;
case event::write: return POLLOUT;
case event::both: return (POLLIN|POLLOUT);
default: CPPA_CRITICAL("invalid event bitmask");
}
}
class middleman_event_handler : public middleman_event_handler_base {
public:
......@@ -519,7 +535,8 @@ class middleman_event_handler : public middleman_event_handler_base {
CPPA_REQUIRE(m_pollset.size() == m_meta.size());
for (;;) {
auto presult = ::poll(m_pollset.data(), m_pollset.size(), -1);
DEBUG("poll() on " << m_pollset.size() << " sockets returned " << presult);
CPPA_LOG_DEBUG("poll() on " << num_sockets()
<< " sockets returned " << presult);
if (presult < 0) {
switch (errno) {
// a signal was caught
......@@ -528,6 +545,7 @@ class middleman_event_handler : public middleman_event_handler_base {
break;
}
case ENOMEM: {
CPPA_LOG_ERROR("poll() failed for reason ENOMEM");
// there's not much we can do other than try again
// in hope someone releases memory
//this_thread::yield();
......@@ -544,69 +562,59 @@ class middleman_event_handler : public middleman_event_handler_base {
}
}
void update() {
auto set_pollfd_events = [](pollfd& pfd, event_bitmask mask) {
switch (mask) {
case event::read: pfd.events = POLLIN; break;
case event::write: pfd.events = POLLOUT; break;
case event::both: pfd.events = (POLLIN|POLLOUT); break;
default: CPPA_CRITICAL("invalid event bitmask");
}
};
// first add, then erase (erase has higher priority)
for (auto& add : m_additions) {
CPPA_REQUIRE((std::get<2>(add) & event::both) != event::none);
auto i = find_if(begin(m_meta), end(m_meta), [&](const fd_meta_info& other) {
return std::get<0>(add) == std::get<0>(other);
});
pollfd* pfd;
event_bitmask mask = std::get<2>(add);
if (i != end(m_meta)) {
CPPA_REQUIRE(std::get<1>(*i) == std::get<1>(add));
mask |= std::get<2>(*i);
pfd = &(m_pollset[distance(begin(m_meta), i)]);
}
else {
protected:
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_reader*) {
static_cast<void>(old_bitmask); // no need for it
switch (me) {
case fd_meta_event::add: {
pollfd tmp;
tmp.fd = std::get<0>(add);
tmp.fd = fd;
tmp.events = to_poll_bitmask(new_bitmask);
tmp.revents = 0;
tmp.events = 0;
m_pollset.push_back(tmp);
m_meta.push_back(add);
pfd = &(m_pollset.back());
m_pollset.insert(lb(fd), tmp);
CPPA_LOG_DEBUG("inserted new element");
break;
}
set_pollfd_events(*pfd, mask);
}
m_additions.clear();
for (auto& sub : m_subtractions) {
CPPA_REQUIRE((std::get<2>(sub) & event::both) != event::none);
auto i = find_if(begin(m_meta), end(m_meta), [&](const fd_meta_info& other) {
return std::get<0>(sub) == std::get<0>(other);
});
if (i != end(m_meta)) {
auto pi = m_pollset.begin();
advance(pi, distance(begin(m_meta), i));
auto mask = std::get<2>(*i) & ~(std::get<2>(sub));
switch (mask) {
case event::none: {
m_meta.erase(i);
m_pollset.erase(pi);
break;
}
default: set_pollfd_events(*pi, mask);
case fd_meta_event::erase: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOG_DEBUG("erased element");
m_pollset.erase(iter);
}
break;
}
case fd_meta_event::mod: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
if (iter != last && iter->fd == fd) {
CPPA_LOG_DEBUG("updated bitmask");
iter->events = to_poll_bitmask(new_bitmask);
}
break;
}
}
m_subtractions.clear();
}
private:
// _ ***
pollfd_vector m_pollset; // \ **
// > these two vectors are always in sync *
fd_meta_vector m_meta; // _/ **
// ***
vector<pollfd>::iterator lb(native_socket_type fd) {
return lower_bound(begin(m_pollset), end(m_pollset), fd, m_pless);
}
vector<pollfd> m_pollset; // always in sync with m_meta
pollfd_less m_pless;
};
......@@ -659,15 +667,15 @@ class middleman_event_handler : public middleman_event_handler_base {
m_events.resize(64);
}
size_t num_sockets() const { return m_epoll_data.size(); }
size_t num_sockets() const { return m_meta.size(); }
pair<event_iterator,event_iterator> poll() {
CPPA_REQUIRE(m_epoll_data.empty() == false);
CPPA_REQUIRE(m_meta.empty() == false);
for (;;) {
DEBUG("epoll_wait on " << m_epoll_data.size() << " sockets");
CPPA_LOG_DEBUG("epoll_wait on " << num_sockets() << " sockets");
auto presult = epoll_wait(m_epollfd, m_events.data(),
(int) m_events.size(), -1);
DEBUG("epoll_wait returned " << presult);
CPPA_LOG_DEBUG("epoll_wait returned " << presult);
if (presult < 0) {
// try again unless critical error occured
presult = 0;
......@@ -692,109 +700,61 @@ class middleman_event_handler : public middleman_event_handler_base {
}
}
void update() {
handle_vec(m_additions, EPOLL_CTL_ADD);
handle_vec(m_subtractions, EPOLL_CTL_DEL);
}
protected:
private:
void handle_vec(vector<fd_meta_info>& vec, int eop) {
for (auto& element : vec) {
auto mask = std::get<2>(element);
CPPA_REQUIRE(mask == event::write || mask == event::read || mask == event::both);
auto ptr = std::get<1>(element).get();
switch (mask) {
case event::read:
epoll_op(eop, ptr->read_handle(), EPOLLIN, ptr);
break;
case event::write: {
CPPA_REQUIRE(ptr->is_peer());
auto dptr = static_cast<peer*>(ptr);
epoll_op(eop, dptr->write_handle(), EPOLLOUT, dptr);
break;
}
case event::both: {
if (ptr->is_peer()) {
auto dptr = static_cast<peer*>(ptr);
auto rd = dptr->read_handle();
auto wr = dptr->write_handle();
if (rd == wr) epoll_op(eop, wr, EPOLLIN|EPOLLOUT, dptr);
else {
epoll_op(eop, rd, EPOLLIN, ptr);
epoll_op(eop, wr, EPOLLOUT, dptr);
}
}
else epoll_op(eop, ptr->read_handle(), EPOLLIN, ptr);
break;
}
default: CPPA_CRITICAL("invalid event mask found in handle_vec");
}
}
vec.clear();
}
// operation: EPOLL_CTL_ADD or EPOLL_CTL_DEL
// mask: EPOLLIN, EPOLLOUT or (EPOLLIN | EPOLLOUT)
void epoll_op(int operation, int fd, uint32_t mask, continuable_reader* ptr) {
CPPA_REQUIRE(ptr != nullptr);
CPPA_REQUIRE(operation == EPOLL_CTL_ADD || operation == EPOLL_CTL_DEL);
CPPA_REQUIRE(mask == EPOLLIN || mask == EPOLLOUT || mask == (EPOLLIN|EPOLLOUT));
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_reader* ptr) {
static_cast<void>(old_bitmask); // no need for it
int operation;
epoll_event ee;
// also fire event on peer shutdown for input operations
ee.events = (mask & EPOLLIN) ? (mask | EPOLLRDHUP) : mask;
ee.data.ptr = ptr;
// check wheter fd is already registered to epoll
auto iter = m_epoll_data.find(fd);
if (iter != end(m_epoll_data)) {
CPPA_REQUIRE(ee.data.ptr == iter->second.data.ptr);
if (operation == EPOLL_CTL_ADD) {
if (mask == iter->second.events) {
// nothing to do here, fd is already registered with
// correct bitmask
return;
}
// modify instead
operation = EPOLL_CTL_MOD;
ee.events |= iter->second.events; // new bitmask
iter->second.events = ee.events; // update bitmask in map
}
else if (operation == EPOLL_CTL_DEL) {
// check wheter we have fd registered for other events
ee.events = iter->second.events & ~(ee.events);
if (ee.events != 0) {
iter->second.events = ee.events; // update bitmask in map
operation = EPOLL_CTL_MOD; // modify instead
}
else m_epoll_data.erase(iter); // erase from map as well
}
}
else if (operation == EPOLL_CTL_DEL) {
// nothing to delete here
return;
switch (new_bitmask) {
case event::none:
CPPA_REQUIRE(me == fd_meta_event::erase);
ee.events = 0;
break;
case event::read:
ee.events = EPOLLIN | EPOLLRDHUP;
break;
case event::write:
ee.events = EPOLLOUT;
case event::both:
ee.events = EPOLLIN | EPOLLRDHUP | EPOLLOUT;
break;
default: CPPA_CRITICAL("invalid event bitmask");
}
else { // operation == EPOLL_CTL_ADD
CPPA_REQUIRE(operation == EPOLL_CTL_ADD);
m_epoll_data.insert(make_pair(fd, ee));
switch (me) {
case fd_meta_event::add:
operation = EPOLL_CTL_ADD;
break;
case fd_meta_event::erase:
operation = EPOLL_CTL_DEL;
break;
case fd_meta_event::mod:
operation = EPOLL_CTL_MOD;
break;
default: CPPA_CRITICAL("invalid fd_meta_event");
}
if (epoll_ctl(m_epollfd, operation, fd, &ee) < 0) {
switch (errno) {
// supplied file descriptor is already registered
case EEXIST: {
// shouldn't happen, but no big deal
cerr << "*** warning: file descriptor registered twice\n"
<< flush;
CPPA_LOG_ERROR("file descriptor registered twice");
break;
}
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance.
case ENOENT: {
cerr << "*** warning: cannot delete file descriptor "
"because it isn't registered\n"
<< flush;
CPPA_LOG_ERROR("cannot delete file descriptor "
"because it isn't registered");
break;
}
default: {
CPPA_LOG_ERROR(strerror(errno));
perror("epoll_ctl() failed");
CPPA_CRITICAL("epoll_ctl() failed");
}
......@@ -804,7 +764,6 @@ class middleman_event_handler : public middleman_event_handler_base {
int m_epollfd;
vector<epoll_event> m_events;
map<native_socket_type,epoll_event> m_epoll_data;
};
......@@ -814,49 +773,36 @@ middleman::middleman() : m_done(false), m_handler(new middleman_event_handler){}
middleman::~middleman() { }
void middleman::register_peer(const process_information& node,
const peer_ptr& ptr) {
auto& ptrref = m_peers[node];
if (ptrref) { DEBUG("peer already defined!"); }
else ptrref = ptr;
void middleman::continue_writer(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
CPPA_REQUIRE(ptr->as_writer() != nullptr);
m_handler->add(ptr, event::write);
}
void middleman::continue_writing_later(const peer_ptr& ptr) {
m_handler->add_later(ptr, event::write);
void middleman::stop_writer(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
CPPA_REQUIRE(ptr->as_writer() != nullptr);
m_handler->erase(ptr, event::write);
}
void middleman::add(const continuable_reader_ptr& what) {
m_readers.push_back(what);
m_handler->add_later(what, event::read);
void middleman::continue_reader(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
m_handler->add(ptr, event::read);
}
void middleman::erase(const continuable_reader_ptr& what) {
m_handler->erase_later(what, event::both);
erase_from(m_readers, what);
erase_from_if(m_peers, [=](const map<process_information,peer_ptr>::value_type& kvp) {
return kvp.second == what;
});
}
continuable_reader_ptr middleman::acceptor_of(const actor_ptr& whom) {
auto e = end(m_readers);
auto i = find_if(begin(m_readers), e, [=](const continuable_reader_ptr& crp) {
return crp->is_acceptor_of(whom);
});
return (i != e) ? *i : nullptr;
}
peer_ptr middleman::get_peer(const process_information& node) {
auto e = end(m_peers);
auto i = m_peers.find(node);
return (i != e) ? i->second : nullptr;
void middleman::stop_reader(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase(ptr, event::read);
erase_from(m_readers, ptr);
}
void middleman_loop(middleman_impl* impl) {
middleman_event_handler* handler = impl->m_handler.get();
DEBUG("run middleman loop");
CPPA_LOGF_TRACE("run middleman loop, node: "
<< to_string(*process_information::get()));
handler->init();
impl->add(new middleman_overseer(impl, impl->m_pipe_read, impl->m_queue));
impl->continue_reader(new middleman_overseer(impl->m_pipe_read, impl->m_queue));
handler->update();
while (!impl->done()) {
auto iters = handler->poll();
......@@ -867,45 +813,51 @@ void middleman_loop(middleman_impl* impl) {
case event::none: break;
case event::both:
case event::write: {
DEBUG("handle event::write");
CPPA_LOGF_DEBUG("handle event::write");
switch (i->continue_writing()) {
case write_closed:
case write_failure:
impl->erase(i->ptr());
impl->stop_writer(i->ptr());
CPPA_LOGF_DEBUG("writer removed because of error");
break;
case write_done:
handler->erase_later(i->ptr(), event::write);
impl->stop_writer(i->ptr());
break;
default: break;
}
if (mask == event::write) break;
// else: fall through
DEBUG("handle event::both; fall through");
CPPA_LOGF_DEBUG("handle event::both; fall through");
}
case event::read: {
DEBUG("handle event::read");
CPPA_LOGF_DEBUG("handle event::read");
switch (i->continue_reading()) {
case read_closed:
case read_failure:
impl->erase(i->ptr());
impl->stop_reader(i->ptr());
CPPA_LOGF_DEBUG("remove peer");
break;
default: break;
}
break;
}
case event::error: {
impl->erase(i->ptr());
// calls handler->erase_later(i->ptr(), event::both);
impl->stop_reader(i->ptr());
impl->stop_writer(i->ptr());
CPPA_LOGF_DEBUG("event::error; remove peer");
}
}
i->handled();
}
handler->update();
}
DEBUG("flush outgoing messages");
CPPA_LOGF_DEBUG("event loop done, erase all readers");
// make sure to write everything before shutting down
for (auto ptr : impl->m_readers) { handler->erase_later(ptr, event::read); }
for (auto ptr : impl->m_readers) { handler->erase(ptr, event::read); }
handler->update();
CPPA_LOGF_DEBUG("flush outgoing messages");
CPPA_LOGF_DEBUG_IF(handler->num_sockets() == 0,
"nothing to flush, no writer left");
while (handler->num_sockets() > 0) {
auto iters = handler->poll();
for (auto i = iters.first; i != iters.second; ++i) {
......@@ -915,18 +867,27 @@ void middleman_loop(middleman_impl* impl) {
case write_closed:
case write_failure:
case write_done:
handler->erase_later(i->ptr(), event::write);
handler->erase(i->ptr(), event::write);
break;
default: break;
}
break;
default: CPPA_CRITICAL("expected event::write only "
"during shutdown phase");
case event::error:
handler->erase(i->ptr(), event::both);
break;
default:
CPPA_LOGF_ERROR("expected event::write only "
"during shutdown phase");
handler->erase(i->ptr(), event::read);
break;
}
}
handler->update();
}
DEBUG("middleman loop done");
CPPA_LOGF_DEBUG("clear all containers");
//impl->m_peers.clear();
impl->m_readers.clear();
CPPA_LOGF_DEBUG("middleman loop done");
}
} } // namespace cppa::detail
......@@ -28,20 +28,38 @@
\******************************************************************************/
#include "cppa/network/protocol.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/peer_acceptor.hpp"
#include "cppa/detail/logging.hpp"
namespace cppa { namespace network {
peer_acceptor::peer_acceptor(middleman* mm, native_socket_type fd, const actor_ptr& pa)
: super(mm, fd, false), m_published_actor(pa) { }
protocol::protocol(middleman* parent) : m_parent(parent) { }
void protocol::run_later(std::function<void()> fun) {
m_parent->run_later([=] { fun(); });
}
void protocol::continue_reader(continuable_reader* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->continue_reader(ptr);
}
void protocol::continue_writer(continuable_reader* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
CPPA_REQUIRE(ptr->as_writer() != nullptr);
m_parent->continue_writer(ptr);
}
bool peer_acceptor::is_acceptor_of(const actor_ptr& whom) const {
return m_published_actor == whom;
void protocol::stop_reader(continuable_reader* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_reader(ptr);
}
void peer_acceptor::add_peer(const peer_ptr& ptr) {
parent()->add(ptr);
void protocol::stop_writer(continuable_reader* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_writer(ptr);
}
} } // namespace cppa::network
......@@ -32,7 +32,8 @@
namespace cppa {
serializer::~serializer() {
}
serializer::serializer(actor_addressing* aa) : m_addressing(aa) { }
serializer::~serializer() { }
} // namespace cppa
......@@ -39,6 +39,7 @@
#include "cppa/network/middleman.hpp"
#include "cppa/detail/logging.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/empty_tuple.hpp"
#include "cppa/detail/group_manager.hpp"
......@@ -66,6 +67,7 @@ std::atomic<actor_registry*> s_actor_registry;
std::atomic<group_manager*> s_group_manager;
std::atomic<empty_tuple*> s_empty_tuple;
std::atomic<scheduler*> s_scheduler;
std::atomic<logging*> s_logger;
template<typename T>
T* lazy_get(std::atomic<T*>& ptr) {
......@@ -90,6 +92,7 @@ namespace cppa { void shutdown() { singleton_manager::shutdown(); } }
namespace cppa { namespace detail {
void singleton_manager::shutdown() {
CPPA_LOGF_DEBUG("prepare to shutdown");
if (self.unchecked() != nullptr) {
try { self.unchecked()->quit(exit_reason::normal); }
catch (actor_exited&) { }
......@@ -98,8 +101,8 @@ void singleton_manager::shutdown() {
if (rptr) {
rptr->await_running_count_equal(0);
}
stop_and_kill(s_middleman);
stop_and_kill(s_scheduler);
stop_and_kill(s_middleman);
std::atomic_thread_fence(std::memory_order_seq_cst);
// it's safe now to delete all other singletons now
delete s_actor_registry.load();
......@@ -113,6 +116,8 @@ void singleton_manager::shutdown() {
s_uniform_type_info_map = nullptr;
delete s_decorated_names_map.load();
s_decorated_names_map = nullptr;
// last but not least: kill logger
stop_and_kill(s_logger);
}
actor_registry* singleton_manager::get_actor_registry() {
......@@ -135,6 +140,21 @@ decorated_names_map* singleton_manager::get_decorated_names_map() {
return lazy_get(s_decorated_names_map);
}
logging* singleton_manager::get_logger() {
auto result = s_logger.load();
if (result == nullptr) {
auto tmp = logging::create_singleton();
if (s_logger.compare_exchange_weak(result, tmp) == false) {
delete tmp;
}
else {
result = tmp;
result->start();
}
}
return result;
}
bool singleton_manager::set_scheduler(scheduler* ptr) {
scheduler* expected = nullptr;
if (s_scheduler.compare_exchange_weak(expected, ptr)) {
......
......@@ -44,6 +44,8 @@
#include "cppa/primitive_variant.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/network/default_actor_addressing.hpp"
using namespace std;
namespace cppa {
......@@ -59,7 +61,10 @@ bool isbuiltin(const string& type_name) {
// - atoms are serialized '...'
class string_serializer : public serializer {
typedef serializer super;
ostream& out;
network::default_actor_addressing m_addressing;
struct pt_writer {
......@@ -108,8 +113,7 @@ class string_serializer : public serializer {
public:
string_serializer(ostream& mout)
: out(mout), m_after_value(false), m_obj_just_opened(false) {
}
: super(&m_addressing), out(mout), m_after_value(false), m_obj_just_opened(false) { }
void begin_object(const string& type_name) {
clear();
......@@ -191,11 +195,14 @@ class string_serializer : public serializer {
class string_deserializer : public deserializer {
typedef deserializer super;
string m_str;
string::iterator m_pos;
//size_t m_obj_count;
stack<bool> m_obj_had_left_parenthesis;
stack<string> m_open_objects;
network::default_actor_addressing m_addressing;
void skip_space_and_comma() {
while (*m_pos == ' ' || *m_pos == ',') ++m_pos;
......@@ -263,11 +270,7 @@ class string_deserializer : public deserializer {
public:
string_deserializer(const string& str) : m_str(str) {
m_pos = m_str.begin();
}
string_deserializer(string&& str) : m_str(move(str)) {
string_deserializer(string str) : super(&m_addressing), m_str(move(str)) {
m_pos = m_str.begin();
}
......@@ -510,4 +513,10 @@ string to_string_impl(const void *what, const uniform_type_info *utype) {
} // namespace detail
string to_verbose_string(const exception& e) {
std::ostringstream oss;
oss << detail::demangle(typeid(e)) << ": " << e.what();
return oss.str();
}
} // namespace cppa
......@@ -54,7 +54,6 @@
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
......@@ -62,49 +61,26 @@ namespace cppa {
using namespace detail;
using namespace network;
void publish(actor_ptr whom, std::unique_ptr<acceptor> acceptor) {
if (!whom && !acceptor) return;
singleton_manager::get_actor_registry()->put(whom->id(), whom);
singleton_manager::get_middleman()->publish(std::move(acceptor), whom);
namespace {
protocol* proto() {
return singleton_manager::get_middleman()->protocol(atom("DEFAULT")).get();
}
} // namespace <anonymous>
actor_ptr remote_actor(io_stream_ptr_pair peer) {
auto pinf = process_information::get();
std::uint32_t process_id = pinf->process_id();
// throws on error
peer.second->write(&process_id, sizeof(std::uint32_t));
peer.second->write(pinf->node_id().data(), pinf->node_id().size());
std::uint32_t remote_actor_id;
std::uint32_t peer_pid;
process_information::node_id_type peer_node_id;
peer.first->read(&remote_actor_id, sizeof(remote_actor_id));
peer.first->read(&peer_pid, sizeof(std::uint32_t));
peer.first->read(peer_node_id.data(), peer_node_id.size());
process_information_ptr pinfptr(new process_information(peer_pid, peer_node_id));
if (*pinf == *pinfptr) {
// dude, this is not a remote actor, it's a local actor!
# ifdef CPPA_DEBUG
std::cerr << "*** warning: remote_actor() called to access a local actor\n"
<< std::flush;
# endif
return singleton_manager::get_actor_registry()->get(remote_actor_id);
}
//auto key = std::make_tuple(remote_actor_id, pinfptr->process_id(), pinfptr->node_id());
singleton_manager::get_middleman()->add_peer(peer, pinfptr);
return get_actor_proxy_cache().get_or_put(remote_actor_id,
pinfptr->process_id(),
pinfptr->node_id());
void publish(actor_ptr whom, std::unique_ptr<acceptor> aptr) {
proto()->publish(whom, move(aptr), {});
}
actor_ptr remote_actor(io_stream_ptr_pair io) {
return proto()->remote_actor(io, {});
}
void publish(actor_ptr whom, std::uint16_t port, const char* addr) {
if (whom) publish(whom, ipv4_acceptor::create(port, addr));
proto()->publish(whom, {port, addr});
}
actor_ptr remote_actor(const char* host, std::uint16_t port) {
// throws on error
io_stream_ptr peer = ipv4_io_stream::connect_to(host, port);
io_stream_ptr_pair ptrpair(peer, peer);
return remote_actor(ptrpair);
return proto()->remote_actor({port, host});
}
} // namespace cppa
......@@ -47,6 +47,7 @@
#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"
......@@ -57,7 +58,6 @@
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
......@@ -175,66 +175,20 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr> {
static void s_serialize(const actor_ptr& ptr,
serializer* sink,
const string& name) {
if (ptr == nullptr) {
serialize_nullptr(sink);
}
else {
if (ptr->is_proxy() == false) {
singleton_manager::get_actor_registry()->put(ptr->id(), ptr);
}
primitive_variant ptup[3];
ptup[0] = ptr->id();
ptup[1] = ptr->parent_process().process_id();
ptup[2] = to_string(ptr->parent_process().node_id());
sink->begin_object(name);
sink->write_tuple(3, ptup);
sink->end_object();
}
const string&) {
auto impl = sink->addressing();
if (impl) impl->write(sink, ptr);
else throw std::runtime_error("unable to serialize actor_ptr: "
"no actor addressing defined");
}
static void s_deserialize(actor_ptr& ptrref,
deserializer* source,
const string& name) {
auto cname = source->seek_object();
if (cname != name) {
if (cname == s_nullptr_type_name) {
deserialize_nullptr(source);
ptrref.reset();
}
else assert_type_name(source, name); // throws
}
else {
primitive_variant ptup[3];
primitive_type ptypes[] = { pt_uint32, pt_uint32, pt_u8string };
source->begin_object(cname);
source->read_tuple(3, ptypes, ptup);
source->end_object();
const string& nstr = get<string>(ptup[2]);
// local actor?
auto pinf = process_information::get();
if ( pinf->process_id() == get<uint32_t>(ptup[1])
&& cppa::equal(nstr, pinf->node_id())) {
auto id = get<uint32_t>(ptup[0]);
ptrref = singleton_manager::get_actor_registry()->get(id);
//ptrref = actor::by_id(get<uint32_t>(ptup[0]));
}
else {
/*
actor_proxy_cache::key_tuple key;
get<0>(key) = get<uint32_t>(ptup[0]);
get<1>(key) = get<uint32_t>(ptup[1]);
node_id_from_string(nstr, get<2>(key));
ptrref = detail::get_actor_proxy_cache().get(key);
*/
process_information::node_id_type nid;
node_id_from_string(nstr, nid);
auto& cache = detail::get_actor_proxy_cache();
ptrref = cache.get_or_put(get<uint32_t>(ptup[0]),
get<uint32_t>(ptup[1]),
nid);
}
}
const string&) {
auto impl = source->addressing();
if (impl) ptrref = impl->read(source);
else throw std::runtime_error("unable to deserialize actor_ptr: "
"no actor addressing defined");
}
protected:
......
......@@ -6,6 +6,8 @@
#include "cppa/sb_actor.hpp"
#include "cppa/to_string.hpp"
#include "cppa/detail/logging.hpp"
using std::cout;
using std::endl;
using namespace cppa;
......@@ -15,35 +17,40 @@ namespace {
size_t s_pongs = 0;
behavior ping_behavior(size_t num_pings) {
return on(atom("pong"), arg_match) >> [num_pings](int value) {
//cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs >= num_pings) {
reply(atom("EXIT"), exit_reason::user_defined);
self->quit();
}
else {
reply(atom("ping"), value);
}
},
others() >> [] {
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
self->quit(exit_reason::user_defined);
};
return (
on(atom("pong"), arg_match) >> [num_pings](int value) {
CPPA_LOGF_ERROR_IF(!self->last_sender(), "last_sender() == nullptr");
//cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs >= num_pings) {
reply(atom("EXIT"), exit_reason::user_defined);
self->quit();
}
else reply(atom("ping"), value);
},
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
<< to_string(self->last_dequeued()));
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
self->quit(exit_reason::user_defined);
}
);
}
behavior pong_behavior() {
return on<atom("ping"), int>() >> [](int value) {
//cout << to_string(self->last_dequeued()) << endl;
reply(atom("pong"), value + 1);
},
others() >> []() {
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
self->quit(exit_reason::user_defined);
};
return (
on<atom("ping"), int>() >> [](int value) {
//cout << to_string(self->last_dequeued()) << endl;
reply(atom("pong"), value + 1);
},
others() >> []() {
cout << "unexpected message; "
<< __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
self->quit(exit_reason::user_defined);
}
);
}
} // namespace <anonymous>
......@@ -53,11 +60,13 @@ size_t pongs() {
}
void ping(size_t num_pings) {
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
receive_loop(ping_behavior(num_pings));
}
actor_ptr spawn_event_based_ping(size_t num_pings) {
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
return factory::event_based(
[num_pings] {
......@@ -67,12 +76,14 @@ actor_ptr spawn_event_based_ping(size_t num_pings) {
}
void pong(actor_ptr ping_actor) {
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
// kickoff
send(ping_actor, atom("pong"), 0);
receive_loop (pong_behavior());
}
actor_ptr spawn_event_based_pong(actor_ptr ping_actor) {
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
CPPA_REQUIRE(ping_actor.get() != nullptr);
return factory::event_based([=] {
self->become(pong_behavior());
......
......@@ -8,6 +8,7 @@
#include "ping_pong.hpp"
#include "cppa/cppa.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/logging.hpp"
using namespace std;
using namespace cppa;
......@@ -110,7 +111,7 @@ void spawn5_server(actor_ptr client, bool inverted) {
default_case,
after(chrono::seconds(2)) >> [&] {
i = 4;
cout << "received timeout while waiting for DOWN messages!\n";
cout << "timeout while waiting for DOWN messages!\n";
}
);}
// wait for locally spawned reflectors
......@@ -217,11 +218,10 @@ int client_part(const vector<string_pair>& args) {
} // namespace <anonymous>
void verbose_terminate() {
try { throw; }
try { if (std::uncaught_exception()); throw; }
catch (std::exception& e) {
cerr << "terminate called after throwing "
<< detail::demangle(typeid(e))
<< ", reason: " << e.what() << endl;
<< to_verbose_string(e) << endl;
}
catch (...) {
cerr << "terminate called after throwing an unknown exception" << endl;
......@@ -246,6 +246,7 @@ int main(int argc, char** argv) {
}
}
CPPA_TEST(test__remote_actor);
CPPA_LOGF_TRACE("");
//auto ping_actor = spawn(ping, 10);
uint16_t port = 4242;
bool success = false;
......@@ -253,6 +254,7 @@ int main(int argc, char** argv) {
try {
publish(self, port, "127.0.0.1");
success = true;
CPPA_LOGF_DEBUG("running on port " << port);
}
catch (bind_failure&) {
// try next port
......@@ -267,6 +269,7 @@ int main(int argc, char** argv) {
// execute client_part() in a separate process,
// connected via localhost socket
child = thread([&oss]() {
CPPA_LOGF_TRACE("lambda expression calling system()");
string cmdstr = oss.str();
if (system(cmdstr.c_str()) != 0) {
cerr << "FATAL: command \"" << cmdstr << "\" failed!" << endl;
......@@ -279,12 +282,16 @@ int main(int argc, char** argv) {
}
//cout << "await SpawnPing message" << endl;
actor_ptr remote_client;
CPPA_LOGF_DEBUG("send 'SpawnPing', expect 'PingPtr'");
receive (
on(atom("SpawnPing")) >> [&]() {
remote_client = self->last_sender();
CPPA_LOGF_ERROR_IF(!remote_client, "last_sender() == nullptr");
CPPA_LOGF_DEBUG("spawn 10 event-based ping actors");
reply(atom("PingPtr"), spawn_event_based_ping(10));
}
);
CPPA_LOGF_DEBUG("wait until spawned ping actors are done");
await_all_others_done();
CPPA_CHECK_EQUAL(10, pongs());
cout << "test remote sync_send" << endl;
......
......@@ -161,9 +161,7 @@ int main() {
CPPA_CHECK_EQUAL(get<1>(tup), "foo");
}
}
catch (exception& e) {
CPPA_ERROR("exception: " << e.what());
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
try {
// test raw_type in both binary and string serialization
......@@ -182,9 +180,7 @@ int main() {
rs2 = from_string<raw_struct>(rsstr);
CPPA_CHECK_EQUAL(rs.str, rs2.str);
}
catch (exception& e) {
CPPA_ERROR("exception: " << e.what());
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
{
auto ttup = make_any_tuple(1, 2, actor_ptr(self));
......@@ -220,9 +216,7 @@ int main() {
CPPA_CHECK_EQUAL(get<1>(tup), "foo");
}
}
catch (exception& e) {
CPPA_ERROR("exception: " << e.what());
}
catch (exception& e) { CPPA_ERROR(to_verbose_string(e)); }
}
{
any_tuple msg1 = cppa::make_cow_tuple(42, string("Hello \"World\"!"));
......
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