Commit c3b37711 authored by Dominik Charousset's avatar Dominik Charousset

Add hook infrastructure to middleman, relates #150

parent a4a1a55f
......@@ -7,3 +7,6 @@
[submodule "nexus"]
path = nexus
url = https://github.com/actor-framework/nexus.git
[submodule "probe"]
path = probe
url = https://github.com/actor-framework/probe.git
......@@ -228,7 +228,7 @@ set(LIBCAF_IO_LIBRARY libcaf_io)
if(CAF_BUILD_STATIC_ONLY OR CAF_BUILD_STATIC)
add_dependencies(libcaf_ioStatic libcaf_coreStatic)
endif()
if(NOT "${CAF_BUILD_STATIC_ONLY}" STREQUAL "yes")
if(NOT CAF_BUILD_STATIC_ONLY)
add_dependencies(libcaf_io libcaf_core)
endif()
# set LIBCAF_LIBRARIES for other subprojects
......@@ -261,6 +261,23 @@ else()
# make sure variable is set for build log
set(CAF_NO_NEXUS yes)
endif()
# build probe if not being told otherwise
if(NOT CAF_NO_PROBE AND EXISTS "${CMAKE_SOURCE_DIR}/probe/caf/")
if(NOT CAF_HAS_PROBE_EVENTS)
message(WARNING "cannot build probe: probe-event submodule missing")
set(CAF_NO_PROBE yes)
else()
message(STATUS "Enter subdirectory probe")
add_subdirectory(probe)
if(CAF_BUILD_STATIC_ONLY OR CAF_BUILD_STATIC)
add_dependencies(libcaf_probeStatic libcaf_ioStatic)
endif()
if(NOT CAF_BUILD_STATIC_ONLY)
add_dependencies(libcaf_probe libcaf_io)
endif()
set(CAF_NO_PROBE no)
endif()
endif()
# build benchmarks if not being told otherwise
if(NOT CAF_NO_BENCHMARKS AND EXISTS "${CMAKE_SOURCE_DIR}/benchmarks/caf/")
message(STATUS "Enter subdirectory benchmarks")
......@@ -341,6 +358,7 @@ message(STATUS
"\nBuild examples: ${CAF_BUILD_EXAMPLES}"
"\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}"
"\nBuild nexus: ${CAF_BUILD_NEXUS}"
"\nBuild probe: ${CAF_BUILD_PROBE}"
"\nBuild benchmarks: ${CAF_BUILD_BENCHMARKS}"
"\n"
"\nCXX: ${CMAKE_CXX_COMPILER}"
......
Subproject commit 5a0d36801a387b718e4bb1b34bcb75f3e6e25d3f
Subproject commit 015c3bcd41098945ba85d2b0c756c9955b14f53c
......@@ -36,6 +36,7 @@ class node_id;
class behavior;
class resumable;
class actor_addr;
class message_id;
class local_actor;
class actor_proxy;
class scoped_actor;
......@@ -55,6 +56,9 @@ struct invalid_actor_addr_t;
// enums
enum class atom_value : uint64_t;
// aliases
using actor_id = uint32_t;
// intrusive pointer types
using abstract_group_ptr = intrusive_ptr<abstract_group>;
using actor_proxy_ptr = intrusive_ptr<actor_proxy>;
......
......@@ -39,7 +39,6 @@ class message_builder {
message_builder();
message_builder(const message_builder&) = delete;
message_builder& operator=(const message_builder&) = delete;
~message_builder();
/**
......
......@@ -72,17 +72,20 @@ inline std::string to_string(const channel& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const message_id& what) {
return detail::to_string_impl(what);
}
// implemented in node_id.cpp
std::string to_string(const node_id& what);
// implemented in node_id.cpp
std::string to_string(const node_id& what);
/*
inline std::string to_string(const any& what) {
return detail::to_string_impl(what.value(), what.type());
template <class T, class U, class... Us>
std::string to_string(const T& a1, const U& a2, const Us&... as) {
return to_string(a1) + ", " + to_string(a2, as...);
}
*/
/**
* Converts `e` to a string including the demangled type of `e` and `e.what()`.
......
......@@ -11,6 +11,7 @@ set (LIBCAF_IO_SRCS
src/broker.cpp
src/max_msg_size.cpp
src/middleman.cpp
src/hook.cpp
src/network.cpp
src/publish_local_groups.cpp
src/remote_actor_proxy.cpp)
......
......@@ -115,54 +115,78 @@ class basp_broker : public broker, public actor_namespace::backend {
void write(binary_serializer& bs, const basp::header& msg);
void send(const connection_context& ctx, const basp::header& msg,
message payload);
message payload);
void send_kill_proxy_instance(const id_type& nid, actor_id aid,
uint32_t reason);
uint32_t reason);
connection_state handle_basp_header(connection_context& ctx,
const buffer_type* payload = nullptr);
const buffer_type* payload = nullptr);
optional<skip_message_t> add_monitor(connection_context& ctx,
actor_id aid);
optional<skip_message_t> add_monitor(connection_context& ctx, actor_id aid);
optional<skip_message_t> kill_proxy(connection_context& ctx,
actor_id aid,
std::uint32_t reason);
optional<skip_message_t> kill_proxy(connection_context& ctx, actor_id aid,
std::uint32_t reason);
void announce_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom);
const abstract_actor_ptr& whom);
void new_data(connection_context& ctx, buffer_type& buf);
void init_handshake_as_client(connection_context& ctx,
client_handshake_data* ptr);
client_handshake_data* ptr);
void init_handshake_as_sever(connection_context& ctx,
actor_addr published_actor);
actor_addr published_actor);
void serialize_msg(const actor_addr& sender, message_id mid,
const message& msg, buffer_type& wr_buf);
const message& msg, buffer_type& wr_buf);
bool try_set_default_route(const id_type& nid, connection_handle hdl);
void add_route(const id_type& nid, connection_handle hdl);
connection_handle get_route(const id_type& dest);
struct connection_info {
connection_handle hdl;
node_id node;
inline bool invalid() const {
return hdl.invalid();
}
inline bool operator==(const connection_info& other) const {
return hdl == other.hdl && node == other.node;
}
inline bool operator<(const connection_info& other) const {
return hdl < other.hdl;
}
};
connection_info get_route(const id_type& dest);
struct connection_info_less {
inline bool operator()(const connection_info& lhs,
const connection_info& rhs) const {
return lhs.hdl < rhs.hdl;
}
inline bool operator()(const connection_info& lhs,
const connection_handle& rhs) const {
return lhs.hdl < rhs;
}
};
using blacklist_entry = std::pair<id_type, connection_handle>;
// (default route, [alternative routes])
using routing_table_entry = std::pair<connection_handle,
std::set<connection_handle>>;
using routing_table_entry = std::pair<connection_info,
std::set<connection_info>>;
struct blacklist_less {
inline bool operator()(const blacklist_entry& lhs,
const blacklist_entry& rhs) const {
if (lhs.first < rhs.first) return lhs.second < rhs.second;
const blacklist_entry& rhs) const {
if (lhs.first < rhs.first) {
return lhs.second < rhs.second;
}
return false;
}
};
// dest => hops
......@@ -179,7 +203,7 @@ class basp_broker : public broker, public actor_namespace::backend {
std::set<blacklist_entry, blacklist_less> m_blacklist; // stores invalidated
// routes
std::set<pending_request> m_pending_requests;
std::map<id_type, connection_handle> m_nodes;
//std::map<id_type, connection_handle> m_nodes;
// needed to keep track to which node we are talking to at the moment
connection_context* m_current_context;
......
......@@ -62,16 +62,16 @@ class broker : public extend<local_actor>::
using buffer_type = std::vector<char>;
/**
* Manages a low-level IO device for the `broker`.
*/
class servant {
friend class broker;
public:
friend class broker;
virtual ~servant();
protected:
virtual void remove_from_broker() = 0;
virtual message disconnect_message() = 0;
......@@ -85,19 +85,17 @@ class broker : public extend<local_actor>::
bool m_disconnected;
broker* m_broker;
};
/**
* Manages a stream.
*/
class scribe : public network::stream_manager, public servant {
using super = servant;
public:
scribe(broker* parent, connection_handle hdl);
~scribe();
scribe(broker* parent, connection_handle hdl);
/**
* Implicitly starts the read loop on first call.
*/
......@@ -114,12 +112,13 @@ class broker : public extend<local_actor>::
*/
virtual void flush() = 0;
inline connection_handle hdl() const { return m_hdl; }
inline connection_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
......@@ -139,20 +138,20 @@ class broker : public extend<local_actor>::
connection_handle m_hdl;
message m_read_msg;
};
/**
* Manages incoming connections.
*/
class doorman : public network::acceptor_manager, public servant {
using super = servant;
public:
doorman(broker* parent, accept_handle hdl);
~doorman();
doorman(broker* parent, accept_handle hdl);
inline accept_handle hdl() const { return m_hdl; }
inline accept_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
......@@ -160,7 +159,6 @@ class broker : public extend<local_actor>::
virtual void launch() = 0;
protected:
void remove_from_broker() override;
message disconnect_message() override;
......@@ -176,12 +174,11 @@ class broker : public extend<local_actor>::
accept_handle m_hdl;
message m_accept_msg;
};
class continuation;
// ... and some helpers need friendship
// a broker needs friends
friend class scribe;
friend class doorman;
friend class continuation;
......@@ -213,7 +210,9 @@ class broker : public extend<local_actor>::
/**
* Returns the number of open connections.
*/
inline size_t num_connections() const { return m_scribes.size(); }
inline size_t num_connections() const {
return m_scribes.size();
}
std::vector<connection_handle> connections() const;
......@@ -225,7 +224,7 @@ class broker : public extend<local_actor>::
using fun_res = decltype(fun(this, hdl, std::forward<Ts>(vs)...));
// prevent warning about unused local type
static_assert(std::is_same<fun_res, fun_res>::value,
"your compiler is lying to you");
"your compiler is lying to you");
auto i = m_scribes.find(hdl);
if (i == m_scribes.end()) {
CAF_LOG_ERROR("invalid handle");
......@@ -234,67 +233,54 @@ class broker : public extend<local_actor>::
auto sptr = i->second;
CAF_REQUIRE(sptr->hdl() == hdl);
m_scribes.erase(i);
return spawn_functor(nullptr,
[sptr](broker* forked) {
sptr->set_broker(forked);
forked->m_scribes.insert(
std::make_pair(sptr->hdl(), sptr));
},
fun, hdl, std::forward<Ts>(vs)...);
return spawn_functor(nullptr, [sptr](broker* forked) {
sptr->set_broker(forked);
forked->m_scribes.insert(
std::make_pair(sptr->hdl(), sptr));
},
fun, hdl, std::forward<Ts>(vs)...);
}
template <class Socket>
connection_handle add_connection(Socket sock) {
CAF_LOG_TRACE("");
class impl : public scribe {
using super = scribe;
public:
impl(broker* parent, Socket&& s)
: super(parent, network::conn_hdl_from_socket(s))
, m_launched(false), m_stream(parent->backend()) {
: scribe(parent, network::conn_hdl_from_socket(s)),
m_launched(false),
m_stream(parent->backend()) {
m_stream.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.configure_read(config);
if (!m_launched) launch();
}
buffer_type& wr_buf() override {
return m_stream.wr_buf();
}
buffer_type& rd_buf() override {
return m_stream.rd_buf();
}
void stop_reading() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.stop_reading();
disconnect();
}
void flush() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.flush(this);
}
void launch() {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_REQUIRE(!m_launched);
m_launched = true;
m_stream.start(this);
}
private:
bool m_launched;
network::stream<Socket> m_stream;
};
intrusive_ptr<impl> ptr{new impl{this, std::move(sock)}};
m_scribes.insert(std::make_pair(ptr->hdl(), ptr));
......@@ -306,17 +292,12 @@ class broker : public extend<local_actor>::
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_REQUIRE(sock.fd() != network::invalid_socket);
class impl : public doorman {
using super = doorman;
public:
impl(broker* parent, SocketAcceptor&& s)
: super(parent, network::accept_hdl_from_socket(s))
, m_acceptor(parent->backend()) {
: doorman(parent, network::accept_hdl_from_socket(s)),
m_acceptor(parent->backend()) {
m_acceptor.init(std::move(s));
}
void new_connection() override {
accept_msg().handle = m_broker->add_connection(
std::move(m_acceptor.accepted_socket()));
......@@ -324,18 +305,15 @@ class broker : public extend<local_actor>::
message_id::invalid,
m_accept_msg);
}
void stop_reading() override {
m_acceptor.stop_reading();
disconnect();
}
void launch() override {
m_acceptor.start(this);
}
private:
network::acceptor<SocketAcceptor> m_acceptor;
};
intrusive_ptr<impl> ptr{new impl{this, std::move(sock)}};
m_doormen.insert(std::make_pair(ptr->hdl(), ptr));
......@@ -344,7 +322,7 @@ class broker : public extend<local_actor>::
}
void enqueue(const actor_addr&, message_id, message,
execution_unit*) override;
execution_unit*) override;
template <class F>
static broker_ptr from(F fun) {
......@@ -358,7 +336,7 @@ class broker : public extend<local_actor>::
template <class F, typename T, class... Ts>
static broker_ptr from(F fun, T&& v, Ts&&... vs) {
return from(std::bind(fun, std::placeholders::_1, std::forward<T>(v),
std::forward<Ts>(vs)...));
std::forward<Ts>(vs)...));
}
/**
......@@ -389,16 +367,14 @@ class broker : public extend<local_actor>::
static constexpr auto exactly = receive_policy_flag::exactly;
void receive_policy(connection_handle hdl,
receive_policy_flag flag,
size_t num_bytes) CAF_DEPRECATED {
void receive_policy(connection_handle hdl, receive_policy_flag flag,
size_t num_bytes) CAF_DEPRECATED {
configure_read(hdl, receive_policy::config{flag, num_bytes});
}
// </backward_compatibility>
protected:
broker();
void cleanup(uint32_t reason) override;
......@@ -409,10 +385,10 @@ class broker : public extend<local_actor>::
bool initialized() const;
/** @endcond */
virtual behavior make_behavior() = 0;
/** @endcond */
inline middleman& parent() {
return m_mm;
}
......@@ -420,7 +396,6 @@ class broker : public extend<local_actor>::
network::multiplexer& backend();
private:
template <class Handle, class T>
static T& by_id(Handle hdl, std::map<Handle, intrusive_ptr<T>>& elements) {
auto i = elements.find(hdl);
......@@ -468,8 +443,9 @@ class broker::functor_based : public extend<broker>::
using super = combined_type;
template <class... Ts>
functor_based(Ts&&... vs)
: super(std::forward<Ts>(vs)...) {}
functor_based(Ts&&... vs) : super(std::forward<Ts>(vs)...) {
// nop
}
~functor_based();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENCE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_HOOK_HPP
#define CAF_IO_HOOK_HPP
#include <memory>
#include <vector>
#include "caf/fwd.hpp"
namespace caf {
namespace io {
class hook;
// safes us some typing for the static dispatching
#define CAF_IO_HOOK_DISPATCH(eventname) \
template <typename... Ts> \
void dispatch(event<eventname>, Ts&&... ts) { \
eventname##_cb(std::forward<Ts>(ts)...); \
}
/**
* @relates hook
*/
using hook_uptr = std::unique_ptr<hook>;
/**
* Interface to define hooks into the IO layer.
*/
class hook {
public:
virtual ~hook();
/**
* Called whenever a message has arrived via the network.
*/
virtual void message_received_cb(const node_id& source,
const actor_addr& from,
const actor_addr& dest,
message_id mid,
const message& msg);
/**
* Called whenever a message has been sent to the network.
* @param from The address of the sending actor.
* @param hop The node in the network we've sent the message to.
* @param dest The address of the receiving actor. Note that the node ID
* of `dest` can differ from `hop` in case we don't
* have a direct connection to `dest_node`.
* @param mid The ID of the message.
* @param payload The message we've sent.
*/
virtual void message_sent_cb(const actor_addr& from, const node_id& hop,
const actor_addr& dest, message_id mid,
const message& payload);
/**
* Called whenever no route for sending a message exists.
*/
virtual void message_sending_failed_cb(const actor_addr& from,
const actor_addr& dest,
message_id mid,
const message& payload);
/**
* Called whenever a message is forwarded to a different node.
*/
virtual void message_forwarded_cb(const node_id& from, const node_id& dest,
std::vector<char>* payload);
/**
* Called whenever no route for a forwarding request exists.
*/
virtual void message_forwarding_failed_cb(const node_id& from,
const node_id& to,
std::vector<char>* payload);
/**
* Called whenever an actor has been published.
*/
virtual void actor_published_cb(const actor_addr& addr, uint16_t port);
/**
* Called whenever a new remote actor appeared.
*/
virtual void new_remote_actor_cb(const actor_addr& addr);
/**
* Called whenever a handshake via a direct TCP connection succeeded.
*/
virtual void new_connection_established_cb(const node_id& node);
/**
* Called whenever a message from or to a yet unknown node was received.
* @param via The node that has sent us the message.
* @param node The newly added entry to the routing table.
*/
virtual void new_route_added_cb(const node_id& via, const node_id& node);
/**
* Called whenever a message was discarded because a remote node
* tried to send a message to an actor ID that could not be found
* in the registry.
*/
virtual void invalid_message_received_cb(const node_id& source,
const actor_addr& sender,
actor_id invalid_dest,
message_id mid,
const message& msg);
/**
* All possible events for IO hooks.
*/
enum event_type {
message_received,
message_sent,
message_forwarded,
message_sending_failed,
message_forwarding_failed,
actor_published,
new_remote_actor,
new_connection_established,
new_route_added,
invalid_message_received
};
/**
* Handles an event by invoking the associated callback.
*/
template <event_type Event, typename... Ts>
void handle(Ts&&... ts) {
dispatch(event<Event>{}, std::forward<Ts>(ts)...);
}
/**
* Forwards an event to the next hook.
*/
template <event_type Event, typename... Ts>
void call_next(Ts&&... ts) {
if (next) {
next->dispatch(event<Event>{}, std::forward<Ts>(ts)...);
}
}
/**
* Intrusive pointer to the next hook. Hooks are stored as a simple,
* singly linked list.
*/
hook_uptr next;
private:
// ------------ convenience interface based on static dispatching ------------
template <event_type Id>
using event = std::integral_constant<event_type, Id>;
CAF_IO_HOOK_DISPATCH(message_received)
CAF_IO_HOOK_DISPATCH(message_sent)
CAF_IO_HOOK_DISPATCH(message_forwarded)
CAF_IO_HOOK_DISPATCH(message_sending_failed)
CAF_IO_HOOK_DISPATCH(message_forwarding_failed)
CAF_IO_HOOK_DISPATCH(actor_published)
CAF_IO_HOOK_DISPATCH(new_remote_actor)
CAF_IO_HOOK_DISPATCH(new_connection_established)
CAF_IO_HOOK_DISPATCH(new_route_added)
CAF_IO_HOOK_DISPATCH(invalid_message_received)
};
} // namespace io
} // namespace caf
#endif // CAF_IO_HOOK_HPP
......@@ -30,6 +30,7 @@
#include "caf/actor_namespace.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/io/hook.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/network.hpp"
......@@ -43,18 +44,28 @@ class middleman : public detail::abstract_singleton {
public:
friend class detail::singletons;
~middleman();
/**
* Get middleman instance.
*/
static middleman* instance();
~middleman();
/**
* Returns the broker associated with `name`.
* Returns the broker associated with `name` or creates a
* new instance of type `Impl`.
*/
template <class Impl>
intrusive_ptr<Impl> get_named_broker(atom_value name);
intrusive_ptr<Impl> get_named_broker(atom_value name) {
auto i = m_named_brokers.find(name);
if (i != m_named_brokers.end()) {
return static_cast<Impl*>(i->second.get());
}
intrusive_ptr<Impl> result{new Impl};
result->launch(true, nullptr);
m_named_brokers.insert(std::make_pair(name, result));
return result;
}
/**
* Adds `bptr` to the list of known brokers.
......@@ -73,44 +84,63 @@ class middleman : public detail::abstract_singleton {
/**
* Returns the IO backend used by this middleman.
*/
inline network::multiplexer& backend() { return m_backend; }
inline network::multiplexer& backend() {
return m_backend;
}
/** @cond PRIVATE */
/**
* Invokes the callback(s) associated with given event.
*/
template <hook::event_type Event, typename... Ts>
void notify(Ts&&... ts) {
if (m_hooks) {
m_hooks->handle<Event>(std::forward<Ts>(ts)...);
}
}
// stops uninitialized instances
void dispose() override;
/**
* Adds a new hook to the middleman.
*/
template<class C, typename... Ts>
void add_hook(Ts&&... args) {
// if only we could move a unique_ptr into a lambda in C++11
auto ptr = new C(std::forward<Ts>(args)...);
run_later([=] {
ptr->next.swap(m_hooks);
m_hooks.reset(ptr);
});
}
/** @cond PRIVATE */
// stops an initialized singleton
// stops the singleton
void stop() override;
// initializes a singleton
// deletes the singleton
void dispose() override;
// initializes the singleton
void initialize() override;
/** @endcond */
private:
// guarded by singleton-getter `instance`
middleman();
network::multiplexer m_backend; // networking backend
network::supervisor* m_supervisor; // keeps backend busy
std::thread m_thread; // runs the backend
// networking backend
network::multiplexer m_backend;
// prevents backend from shutting down unless explicitly requested
network::supervisor* m_supervisor;
// runs the backend
std::thread m_thread;
// keeps track of "singleton-like" brokers
std::map<atom_value, broker_ptr> m_named_brokers;
// keeps track of anonymous brokers
std::set<broker_ptr> m_brokers;
// user-defined hooks
hook_uptr m_hooks;
};
template <class Impl>
intrusive_ptr<Impl> middleman::get_named_broker(atom_value name) {
auto i = m_named_brokers.find(name);
if (i != m_named_brokers.end()) return static_cast<Impl*>(i->second.get());
intrusive_ptr<Impl> result{new Impl};
result->launch(true, nullptr);
m_named_brokers.insert(std::make_pair(name, result));
return result;
}
} // namespace io
} // namespace caf
......
......@@ -70,12 +70,16 @@ behavior basp_broker::make_behavior() {
std::vector<id_type> lost_connections;
for (auto& kvp : m_routes) {
auto& entry = kvp.second;
if (entry.first == msg.handle) {
CAF_LOG_DEBUG("lost direct connection to "
<< to_string(kvp.first));
entry.first = invalid_connection_handle;
if (entry.first.hdl == msg.handle) {
CAF_LOG_DEBUG("lost direct connection to " << to_string(kvp.first));
entry.first.hdl = invalid_connection_handle;
}
auto last = entry.second.end();
auto i = std::lower_bound(entry.second.begin(), last, msg.handle,
connection_info_less{});
if (i != last && i->hdl == msg.handle) {
entry.second.erase(i);
}
entry.second.erase(msg.handle);
if (entry.first.invalid() && entry.second.empty()) {
lost_connections.push_back(kvp.first);
}
......@@ -141,12 +145,12 @@ void basp_broker::new_data(connection_context& ctx, buffer_type& buf) {
return;
}
ctx.state = next_state;
configure_read(ctx.hdl, receive_policy::exactly( next_state == await_payload
? ctx.hdr.payload_len
: basp::header_size));
configure_read(ctx.hdl, receive_policy::exactly(next_state == await_payload
? ctx.hdr.payload_len
: basp::header_size));
}
void basp_broker::dispatch(const basp::header& hdr, message&& payload) {
void basp_broker::dispatch(const basp::header& hdr, message&& msg) {
// TODO: provide hook API to allow ActorShell to
// intercept/trace/log each message
actor_addr src;
......@@ -165,13 +169,16 @@ void basp_broker::dispatch(const basp::header& hdr, message&& payload) {
auto dest = singletons::get_actor_registry()->get(hdr.dest_actor);
auto mid = message_id::from_integer_value(hdr.operation_data);
if (!dest) {
CAF_LOG_DEBUG(
"received a message for an invalid actor; "
"could not find an actor with ID "
<< hdr.dest_actor);
CAF_LOG_DEBUG("received a message for an invalid actor; "
"could not find an actor with ID "
<< hdr.dest_actor);
parent().notify<hook::invalid_message_received>(hdr.source_node, src,
hdr.dest_actor, mid, msg);
return;
}
dest->enqueue(src, mid, std::move(payload), nullptr);
parent().notify<hook::message_received>(hdr.source_node, src,
dest->address(), mid, msg);
dest->enqueue(src, mid, std::move(msg), nullptr);
}
void basp_broker::dispatch(const actor_addr& from, const actor_addr& to,
......@@ -181,13 +188,14 @@ void basp_broker::dispatch(const actor_addr& from, const actor_addr& to,
<< CAF_TARG(to, to_string) << ", " << CAF_TARG(msg, to_string));
CAF_REQUIRE(to != nullptr);
auto dest = to.node();
auto hdl = get_route(dest);
if (hdl.invalid()) {
auto route = get_route(dest);
if (route.invalid()) {
parent().notify<hook::message_sending_failed>(from, to, mid, msg);
CAF_LOG_INFO("unable to dispatch message: no route to "
<< to_string(dest) << ", message: " << to_string(msg));
return;
}
auto& buf = wr_buf(hdl);
auto& buf = wr_buf(route.hdl);
// reserve space in the buffer to write the broker message later on
auto wr_pos = buf.size();
char placeholder[basp::header_size];
......@@ -209,7 +217,8 @@ void basp_broker::dispatch(const actor_addr& from, const actor_addr& to,
from.id(), to.id(),
static_cast<uint32_t>(buf.size() - before), basp::dispatch_message,
mid.integer_value()});
flush(hdl);
flush(route.hdl);
parent().notify<hook::message_sent>(from, route.node, to, mid, msg);
}
void basp_broker::read(binary_deserializer& bd, basp::header& msg) {
......@@ -250,18 +259,18 @@ basp_broker::handle_basp_header(connection_context& ctx,
// forward message if not addressed to us; invalid dest_node implies
// that msg is a server_handshake
if (hdr.dest_node != invalid_node_id && hdr.dest_node != node()) {
auto hdl = get_route(hdr.dest_node);
if (hdl.invalid()) {
auto route = get_route(hdr.dest_node);
if (route.invalid()) {
// TODO: signalize that we don't have route to given node
CAF_LOG_INFO("message dropped: no route to node "
<< to_string(hdr.dest_node));
return close_connection;
}
auto& buf = wr_buf(hdl);
auto& buf = wr_buf(route.hdl);
binary_serializer bs{std::back_inserter(buf), &m_namespace};
write(bs, hdr);
if (payload) buf.insert(buf.end(), payload->begin(), payload->end());
flush(hdl);
flush(route.hdl);
return await_header;
}
// handle a message that is addressed to us
......@@ -322,8 +331,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
return close_connection;
}
else if (!try_set_default_route(ctx.remote_id, ctx.hdl)) {
CAF_LOG_INFO("multiple incoming connections "
"from the same node");
CAF_LOG_INFO("multiple incoming connections from the same node");
return close_connection;
}
break;
......@@ -335,8 +343,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
return close_connection;
}
if (hdr.operation_data != basp::version) {
CAF_LOG_INFO("tried to connect to a node with "
"different BASP version");
CAF_LOG_INFO("tried to connect to a node with different BASP version");
return close_connection;
}
ctx.remote_id = hdr.source_node;
......@@ -429,31 +436,31 @@ basp_broker::handle_basp_header(connection_context& ctx,
void basp_broker::send_kill_proxy_instance(const id_type& nid, actor_id aid,
uint32_t reason) {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid) << CAF_ARG(reason));
auto hdl = get_route(nid);
CAF_LOG_DEBUG(CAF_MARG(hdl, id));
if (hdl.invalid()) {
auto route = get_route(nid);
CAF_LOG_DEBUG(CAF_MARG(route.hdl, id));
if (route.invalid()) {
CAF_LOG_INFO("message dropped, no route to node: " << to_string(nid));
return;
}
auto& buf = wr_buf(hdl);
auto& buf = wr_buf(route.hdl);
binary_serializer bs(std::back_inserter(buf), &m_namespace);
write(bs,
{node(), nid, aid, invalid_actor_id,
0, basp::kill_proxy_instance, uint64_t{reason}});
flush(hdl);
flush(route.hdl);
}
connection_handle basp_broker::get_route(const id_type& dest) {
connection_handle hdl;
basp_broker::connection_info basp_broker::get_route(const id_type& dest) {
connection_info res;
auto i = m_routes.find(dest);
if (i != m_routes.end()) {
auto& entry = i->second;
hdl = entry.first;
if (hdl.invalid() && !entry.second.empty()) {
hdl = *entry.second.begin();
res = entry.first;
if (res.invalid() && !entry.second.empty()) {
res = *entry.second.begin();
}
}
return hdl;
return res;
}
actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
......@@ -470,8 +477,8 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
}
// we need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node
auto hdl = get_route(nid);
if (hdl.invalid()) {
auto route = get_route(nid);
if (route.invalid()) {
// this happens if and only if we don't have a path to `nid`
// and m_current_context->hdl has been blacklisted
CAF_LOG_INFO("cannot create a proxy instance for an actor "
......@@ -489,9 +496,11 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
});
});
// tell remote side we are monitoring this actor now
binary_serializer bs(std::back_inserter(wr_buf(hdl)), &m_namespace);
binary_serializer bs(std::back_inserter(wr_buf(route.hdl)), &m_namespace);
write(bs, {node(), nid, invalid_actor_id, aid,
0, basp::announce_proxy_instance, 0});
// run hooks
parent().notify<hook::new_remote_actor>(res->address());
return res;
}
......@@ -507,7 +516,7 @@ void basp_broker::erase_proxy(const id_type& nid, actor_id aid) {
void basp_broker::add_route(const id_type& nid, connection_handle hdl) {
if (m_blacklist.count(std::make_pair(nid, hdl)) == 0) {
m_routes[nid].second.insert(hdl);
m_routes[nid].second.insert({hdl, nid});
}
}
......@@ -518,7 +527,7 @@ bool basp_broker::try_set_default_route(const id_type& nid,
if (entry.first.invalid()) {
CAF_LOG_DEBUG("new default route: " << to_string(nid) << " -> "
<< hdl.id());
entry.first = hdl;
entry.first = {hdl, nid};
return true;
}
return false;
......@@ -554,8 +563,9 @@ void basp_broker::init_handshake_as_sever(connection_context& ctx,
bs1 << addr.id();
auto sigs = addr.message_types();
bs1 << static_cast<uint32_t>(sigs.size());
for (auto& sig : sigs)
for (auto& sig : sigs) {
bs1 << sig;
}
}
// fill padded region with the actual broker message
binary_serializer bs2(buf.begin() + wrpos, &m_namespace);
......
......@@ -74,7 +74,7 @@ void broker::servant::disconnect() {
}
broker::scribe::scribe(broker* parent, connection_handle hdl)
: super(parent), m_hdl(hdl) {
: servant(parent), m_hdl(hdl) {
std::vector<char> tmp;
m_read_msg = make_message(new_data_msg{m_hdl, std::move(tmp)});
}
......@@ -112,7 +112,7 @@ void broker::scribe::io_failure(network::operation op) {
}
broker::doorman::doorman(broker* parent, accept_handle hdl)
: super(parent), m_hdl(hdl) {
: servant(parent), m_hdl(hdl) {
auto hdl2 = connection_handle::from_int(-1);
m_accept_msg = make_message(new_connection_msg{m_hdl, hdl2});
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENCE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/hook.hpp"
#include "caf/message_id.hpp"
namespace caf {
namespace io {
hook::~hook() {
// nop
}
void hook::message_received_cb(const node_id& source, const actor_addr& from,
const actor_addr& dest, message_id mid,
const message& msg) {
call_next<message_received>(source, from, dest, mid, msg);
}
void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node,
const actor_addr& dest, message_id mid,
const message& payload) {
call_next<message_sent>(from, dest_node, dest, mid, payload);
}
void hook::message_forwarded_cb(const node_id& from, const node_id& dest,
std::vector<char>* payload) {
call_next<message_forwarded>(from, dest, payload);
}
void hook::message_sending_failed_cb(const actor_addr& from,
const actor_addr& dest,
message_id mid,
const message& payload) {
call_next<message_sending_failed>(from, dest, mid, payload);
}
void hook::message_forwarding_failed_cb(const node_id& from, const node_id& to,
std::vector<char>* payload) {
call_next<message_forwarding_failed>(from, to, payload);
}
void hook::actor_published_cb(const actor_addr& addr, uint16_t port) {
call_next<actor_published>(addr, port);
}
void hook::new_remote_actor_cb(const actor_addr& addr) {
call_next<new_remote_actor>(addr);
}
void hook::new_connection_established_cb(const node_id& node) {
call_next<new_connection_established>(node);
}
void hook::new_route_added_cb(const node_id& via, const node_id& node) {
call_next<new_route_added>(via, node);
}
void hook::invalid_message_received_cb(const node_id& source,
const actor_addr& sender,
actor_id invalid_dest,
message_id mid,
const message& msg) {
call_next<invalid_message_received>(source, sender, invalid_dest, mid, msg);
}
} // namespace io
} // namespace caf
Subproject commit ce3ac3c6703ebbdbfc130123ed7545fdbb777d06
Subproject commit 7592f1b6b8152445b2b7343d4d67a070001d742a
Subproject commit d5c98c59198de67b9273fba39c54c660f8f2a378
......@@ -373,8 +373,8 @@ void test_remote_actor(std::string app_path, bool run_remote_actor) {
int main(int argc, char** argv) {
announce<actor_vector>();
cout << "this node is: "
<< to_string(caf::detail::singletons::get_node_id()) << endl;
cout << "this node is: " << to_string(caf::detail::singletons::get_node_id())
<< endl;
message_builder{argv + 1, argv + argc}.apply({
on("-c", spro<uint16_t>)>> [](uint16_t port) {
CAF_LOGF_INFO("run in client mode");
......@@ -395,15 +395,12 @@ int main(int argc, char** argv) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
},
on("-s") >> [&] {
},
on("-s") >> [&] {
CAF_PRINT("don't run remote actor (server mode)");
test_remote_actor(argv[0], false);
},
on() >> [&] {
test_remote_actor(argv[0], true);
},
others() >> [&] {
},
on() >> [&] { test_remote_actor(argv[0], true); }, others() >> [&] {
CAF_PRINTERR("usage: " << argv[0] << " [-s|-c PORT]");
}
});
......
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