Commit 91cf5386 authored by Dominik Charousset's avatar Dominik Charousset

Improve logging of IO classes and shutdown process

parent bb7b5e76
......@@ -187,8 +187,6 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
caf::detail::singletons::get_logger()->set_aid(aid_arg)
#endif
#define CAF_CLASS_NAME typeid(*this).name()
#define CAF_PRINT0(lvlname, classname, funname, msg) \
CAF_LOG_IMPL(lvlname, classname, funname, msg)
......@@ -254,7 +252,8 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
* @def CAF_LOGMF
* Logs a message inside a member function.
*/
#define CAF_LOGMF(level, msg) CAF_LOGC(level, CAF_CLASS_NAME, __func__, msg)
#define CAF_LOGMF(level, msg) \
CAF_LOGC(level, typeid(*this).name(), __func__, msg)
/**
* @def CAF_LOGC
......@@ -276,7 +275,7 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
* Logs a message inside a member function.
*/
#define CAF_LOGMF_IF(stmt, level, msg) \
CAF_LOGC_IF(stmt, level, CAF_CLASS_NAME, __func__, msg)
CAF_LOGC_IF(stmt, level, typeid(*this).name(), __func__, msg)
// convenience macros to safe some typing when printing arguments
#define CAF_ARG(arg) #arg << " = " << arg
......@@ -300,31 +299,6 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
#define CAF_LOG_INFO_IF(stmt, msg) CAF_LOGMF_IF(stmt, CAF_INFO, msg)
#define CAF_LOG_TRACE_IF(stmt, msg) CAF_LOGMF_IF(stmt, CAF_TRACE, msg)
#define CAF_LOGC_ERROR(cname, fun, msg) CAF_LOGC(CAF_ERROR, cname, fun, msg)
#define CAF_LOGC_WARNING(cname, fun, msg) CAF_LOGC(CAF_WARNING, cname, fun, msg)
#define CAF_LOGC_DEBUG(cname, fun, msg) CAF_LOGC(CAF_DEBUG, cname, fun, msg)
#define CAF_LOGC_INFO(cname, fun, msg) CAF_LOGC(CAF_INFO, cname, fun, msg)
#define CAF_LOGC_TRACE(cname, fun, msg) CAF_LOGC(CAF_TRACE, cname, fun, msg)
#define CAF_LOGC_ERROR_IF(stmt, cname, fun, msg) \
CAF_LOGC_IF(stmt, CAF_ERROR, cname, fun, msg)
#define CAF_LOGC_WARNING_IF(stmt, cname, fun, msg) \
CAF_LOGC_IF(stmt, CAF_WARNING, cname, fun, msg)
#define CAF_LOGC_DEBUG_IF(stmt, cname, fun, msg) \
CAF_LOGC_IF(stmt, CAF_DEBUG, cname, fun, msg)
#define CAF_LOGC_INFO_IF(stmt, cname, fun, msg) \
CAF_LOGC_IF(stmt, CAF_INFO, cname, fun, msg)
#define CAF_LOGC_TRACE_IF(stmt, cname, fun, msg) \
CAF_LOGC_IF(stmt, CAF_TRACE, cname, fun, msg)
#define CAF_LOGF_ERROR(msg) CAF_LOGF(CAF_ERROR, msg)
#define CAF_LOGF_WARNING(msg) CAF_LOGF(CAF_WARNING, msg)
#define CAF_LOGF_DEBUG(msg) CAF_LOGF(CAF_DEBUG, msg)
......@@ -337,29 +311,4 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
#define CAF_LOGF_INFO_IF(stmt, msg) CAF_LOGF_IF(stmt, CAF_INFO, msg)
#define CAF_LOGF_TRACE_IF(stmt, msg) CAF_LOGF_IF(stmt, CAF_TRACE, msg)
#define CAF_LOGM_ERROR(cname, msg) CAF_LOGC(CAF_ERROR, cname, __func__, msg)
#define CAF_LOGM_WARNING(cname, msg) CAF_LOGC(CAF_WARNING, cname, msg)
#define CAF_LOGM_DEBUG(cname, msg) CAF_LOGC(CAF_DEBUG, cname, __func__, msg)
#define CAF_LOGM_INFO(cname, msg) CAF_LOGC(CAF_INFO, cname, msg)
#define CAF_LOGM_TRACE(cname, msg) CAF_LOGC(CAF_TRACE, cname, __func__, msg)
#define CAF_LOGM_ERROR_IF(stmt, cname, msg) \
CAF_LOGC_IF(stmt, CAF_ERROR, cname, __func__, msg)
#define CAF_LOGM_WARNING_IF(stmt, cname, msg) \
CAF_LOGC_IF(stmt, CAF_WARNING, cname, msg)
#define CAF_LOGM_DEBUG_IF(stmt, cname, msg) \
CAF_LOGC_IF(stmt, CAF_DEBUG, cname, __func__, msg)
#define CAF_LOGM_INFO_IF(stmt, cname, msg) \
CAF_LOGC_IF(stmt, CAF_INFO, cname, __func__, msg)
#define CAF_LOGM_TRACE_IF(stmt, cname, msg) \
CAF_LOGC_IF(stmt, CAF_TRACE, cname, __func__, msg)
#endif // CAF_LOGGING_HPP
......@@ -22,6 +22,8 @@
#include <utility>
#include "caf/detail/logging.hpp"
namespace caf {
namespace detail {
......@@ -30,38 +32,61 @@ class singletons;
// a mixin for simple singleton classes
template <class Derived, class Base = void>
class singleton_mixin : public Base {
private:
friend class singletons;
inline static Derived* create_singleton() { return new Derived; }
inline void dispose() { delete this; }
inline void stop() { }
inline void initialize() {}
static Derived* create_singleton() {
return new Derived;
}
protected:
void dispose() {
delete this;
}
template <class... Ts>
singleton_mixin(Ts&&... args)
: Base(std::forward<Ts>(args)...) {}
void stop() {
CAF_LOG_TRACE("");
}
virtual ~singleton_mixin() {}
void initialize() {
CAF_LOG_TRACE("");
}
protected:
template <class... Ts>
singleton_mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) {
// nop
}
virtual ~singleton_mixin() {
CAF_LOG_TRACE("");
}
};
template <class Derived>
class singleton_mixin<Derived, void> {
private:
friend class singletons;
inline static Derived* create_singleton() { return new Derived; }
inline void dispose() { delete this; }
inline void stop() { }
inline void initialize() { }
static Derived* create_singleton() {
return new Derived;
}
protected:
void dispose() {
delete this;
}
virtual ~singleton_mixin() { }
void stop() {
CAF_LOG_TRACE("");
}
void initialize() {
CAF_LOG_TRACE("");
}
protected:
virtual ~singleton_mixin() {
CAF_LOG_TRACE("");
}
};
} // namespace detail
......
......@@ -81,7 +81,7 @@ class uniform_type_info_map {
inline void dispose() { delete this; }
inline void stop() { }
void stop();
virtual void initialize() = 0;
......
......@@ -53,11 +53,7 @@ constexpr invalid_node_id_t invalid_node_id = invalid_node_id_t{};
*/
class node_id : detail::comparable<node_id>,
detail::comparable<node_id, invalid_node_id_t> {
using super = ref_counted;
public:
node_id() = default;
node_id(const invalid_node_id_t&);
......@@ -80,14 +76,14 @@ class node_id : detail::comparable<node_id>,
* A reference counted container for host ID and process ID.
*/
class data : public ref_counted {
public:
// for singleton API
inline void stop() { }
void stop();
// for singleton API
inline void dispose() { deref(); }
inline void dispose() {
deref();
}
// for singleton API
inline void initialize() {
......@@ -107,7 +103,6 @@ class node_id : detail::comparable<node_id>,
uint32_t process_id;
host_id_type host_id;
};
~node_id();
......@@ -154,9 +149,7 @@ class node_id : detail::comparable<node_id>,
/** @endcond */
private:
intrusive_ptr<data> m_data;
};
} // namespace caf
......
......@@ -162,14 +162,13 @@ class invoke_policy {
template <class Actor, class Fun, class MaybeResponseHdl = int>
optional<message> invoke_fun(Actor* self, Fun& fun,
MaybeResponseHdl hdl = MaybeResponseHdl{}) {
auto mid = self->current_mailbox_element()->mid;
auto res = fun(self->current_mailbox_element()->msg);
# ifdef CAF_LOG_LEVEL
auto msg_str = res ? to_string(*res) : "none";
auto msg = to_string(self->current_mailbox_element()->msg);
# endif
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << "m, msg = " << msg_str);
CAF_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str);
CAF_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str);
auto mid = self->current_mailbox_element()->mid;
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", " << CAF_ARG(msg));
auto res = fun(self->current_mailbox_element()->msg);
CAF_LOG_DEBUG(msg << " => " << to_string(res));
if (!res) {
return none;
}
......
......@@ -38,6 +38,8 @@
#include "caf/policy/profiled.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/detail/logging.hpp"
namespace caf {
namespace scheduler {
......@@ -173,6 +175,7 @@ class profiled_coordinator : public coordinator<Policy> {
}
void stop() override {
CAF_LOG_TRACE("");
super::stop();
auto now = clock_type::now().time_since_epoch();
auto wallclock = m_system_start + (now - m_clock_start);
......
......@@ -54,8 +54,7 @@ class worker : public execution_unit {
CAF_REQUIRE(m_this_thread.get_id() == std::thread::id{});
auto this_worker = this;
m_this_thread = std::thread{[this_worker] {
CAF_LOGC_TRACE("caf::scheduler::worker", "start$lambda",
"id = " << this_worker->id());
CAF_LOGF_TRACE("id = " << this_worker->id());
this_worker->run();
}};
}
......
......@@ -32,16 +32,34 @@ class exception;
namespace caf {
/**
* @relates message
*/
std::string to_string(const message& what);
/**
* @relates group
*/
std::string to_string(const group& what);
/**
* @relates channel
*/
std::string to_string(const channel& what);
/**
* @relates message_id
*/
std::string to_string(const message_id& what);
/**
* @relates actor_addr
*/
std::string to_string(const actor_addr& what);
/**
* @relates actor
*/
std::string to_string(const actor& what);
/**
......@@ -50,10 +68,21 @@ std::string to_string(const actor& what);
std::string to_string(const node_id& what);
/**
* Returns `what` as a string representation.
* @relates atom_value
*/
std::string to_string(const atom_value& what);
/**
* @relates optional
*/
template <class T>
std::string to_string(const optional<T>& what) {
if (!what) {
return "none";
}
return to_string(*what);
}
/**
* Converts `e` to a string including `e.what()`.
*/
......
......@@ -206,6 +206,7 @@ abstract_coordinator* abstract_coordinator::create_singleton() {
}
void abstract_coordinator::initialize() {
CAF_LOG_TRACE("");
// launch utility actors
m_timer = spawn<timer_actor, hidden + detached + blocking_api>();
m_printer = spawn<hidden + detached + blocking_api>(printer_loop);
......
......@@ -52,7 +52,7 @@ abstract_group::module::module(std::string mname) : m_name(std::move(mname)) {
}
void abstract_group::module::stop() {
// nop
CAF_LOG_TRACE("");
}
const std::string& abstract_group::module::name() {
......
......@@ -131,6 +131,7 @@ class local_group : public abstract_group {
void serialize(serializer* sink);
void stop() override {
CAF_LOG_TRACE("");
anon_send_exit(m_broker, exit_reason::user_shutdown);
m_latch.wait();
}
......@@ -169,22 +170,19 @@ class local_broker : public event_based_actor {
behavior make_behavior() override {
return {
on(atom("JOIN"), arg_match) >> [=](const actor& other) {
CAF_LOGC_TRACE("caf::local_broker", "init$JOIN",
CAF_TARG(other, to_string));
CAF_LOG_TRACE(CAF_TSARG(other));
if (other && m_acquaintances.insert(other).second) {
monitor(other);
}
},
on(atom("LEAVE"), arg_match) >> [=](const actor& other) {
CAF_LOGC_TRACE("caf::local_broker", "init$LEAVE",
CAF_TARG(other, to_string));
CAF_LOG_TRACE(CAF_TSARG(other));
if (other && m_acquaintances.erase(other) > 0) {
demonitor(other);
}
},
on(atom("_Forward"), arg_match) >> [=](const message& what) {
CAF_LOGC_TRACE("caf::local_broker", "init$FORWARD",
CAF_TARG(what, to_string));
CAF_LOG_TRACE(CAF_TSARG(what));
// local forwarding
m_group->send_all_subscribers(current_sender(), what, host());
// forward to all acquaintances
......@@ -192,8 +190,7 @@ class local_broker : public event_based_actor {
},
[=](const down_msg&) {
auto sender = current_sender();
CAF_LOGC_TRACE("caf::local_broker", "init$DOWN",
CAF_TARG(sender, to_string));
CAF_LOG_TRACE(CAF_TSARG(sender));
if (sender) {
auto first = m_acquaintances.begin();
auto last = m_acquaintances.end();
......@@ -207,8 +204,7 @@ class local_broker : public event_based_actor {
},
others >> [=] {
auto msg = current_message();
CAF_LOGC_TRACE("caf::local_broker", "init$others",
CAF_TARG(msg, to_string));
CAF_LOG_TRACE(CAF_TSARG(msg));
send_to_acquaintances(msg);
}
};
......@@ -282,6 +278,7 @@ class local_group_proxy : public local_group {
}
void stop() override {
CAF_LOG_TRACE("");
anon_send_exit(m_proxy_broker, exit_reason::user_shutdown);
m_latch.wait();
}
......@@ -378,6 +375,7 @@ class local_group_module : public abstract_group::module {
}
void stop() override {
CAF_LOG_TRACE("");
std::map<std::string, local_group_ptr> imap;
std::map<actor, local_group_ptr> pmap;
{ // critical section
......@@ -427,6 +425,7 @@ std::atomic<size_t> m_ad_hoc_id;
} // namespace <anonymous>
void group_manager::stop() {
CAF_LOG_TRACE("");
modules_map mm;
{ // critical section
std::lock_guard<std::mutex> guard(m_mmap_mtx);
......
......@@ -164,8 +164,7 @@ void local_actor::cleanup(uint32_t reason) {
}
void local_actor::quit(uint32_t reason) {
CAF_LOG_TRACE("reason = " << reason << ", class = "
<< CAF_CLASS_NAME);
CAF_LOG_TRACE(CAF_ARG(reason));
planned_exit_reason(reason);
if (is_blocking()) {
throw actor_exited(reason);
......
......@@ -129,7 +129,8 @@ class logging_impl : public logging {
auto strip_magic = [&](const char* prefix_begin, const char* prefix_end) {
auto last = class_name.end();
auto i = std::search(class_name.begin(), last, prefix_begin, prefix_end);
auto e = std::find(i, last, ',');
auto comma_or_angle_bracket = [](char c) { return c == ',' || c == '>'; };
auto e = std::find_if(i, last, comma_or_angle_bracket);
if (i != e) {
std::string substr(i + (prefix_end - prefix_begin), e);
class_name.swap(substr);
......@@ -142,8 +143,6 @@ class logging_impl : public logging {
# else
std::string class_name = c_class_name;
# endif
replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymous namespace)", "$anon$");
std::string file_name;
std::string full_file_name = c_full_file_name;
auto ri = find(full_file_name.rbegin(), full_file_name.rend(), '/');
......
......@@ -31,6 +31,7 @@
#include "caf/serializer.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp"
......@@ -162,8 +163,13 @@ node_id::data::~data() {
// nop
}
void node_id::data::stop() {
CAF_LOG_TRACE("");
}
// initializes singleton
node_id::data* node_id::data::create_singleton() {
CAF_LOGF_TRACE("");
auto ifs = detail::get_mac_addresses();
std::vector<std::string> macs;
macs.reserve(ifs.size());
......
......@@ -83,14 +83,20 @@ void singletons::stop_singletons() {
stop(s_uniform_type_info_map);
stop(s_node_id);
// dispose singletons, i.e., release memory
CAF_LOGF_DEBUG("dispose scheduler");
dispose(s_scheduling_coordinator);
CAF_LOGF_DEBUG("dispose plugins");
for (auto& plugin : s_plugins) {
dispose(plugin);
}
CAF_LOGF_DEBUG("dispose registry");
dispose(s_actor_registry);
CAF_LOGF_DEBUG("dispose group manager");
dispose(s_group_manager);
CAF_LOGF_DEBUG("dispose map");
dispose(s_uniform_type_info_map);
// final steps
CAF_LOGF_DEBUG("stop and dispose logger, bye");
stop(s_logger);
dispose(s_logger);
dispose(s_node_id);
......
......@@ -77,7 +77,7 @@ const uniform_type_info* uniform_type_info::from(const std::type_info& tinf) {
std::string error = "uniform_type_info::by_type_info(): ";
error += tinf.name();
error += " has not been announced";
CAF_LOGM_ERROR("caf::uniform_type_info", error);
CAF_LOGF_ERROR(error);
throw std::runtime_error(error);
}
return result;
......@@ -113,7 +113,7 @@ const uniform_type_info* uniform_typeid(const std::type_info& tinf,
std::string error = "uniform_typeid(): ";
error += tinf.name();
error += " has not been announced";
CAF_LOGM_ERROR("caf::uniform_type_info", error);
CAF_LOGF_ERROR(error);
throw std::runtime_error(error);
}
......
......@@ -777,6 +777,10 @@ uniform_type_info_map* uniform_type_info_map::create_singleton() {
return new utim_impl;
}
void uniform_type_info_map::stop() {
CAF_LOG_TRACE("");
}
uniform_type_info_map::~uniform_type_info_map() {
// nop
}
......
......@@ -357,8 +357,7 @@ class default_multiplexer : public multiplexer {
// read handle which is only registered for reading
auto old_bf = ptr ? ptr->eventbf() : input_mask;
//auto bf = fun(op, old_bf);
CAF_LOG_TRACE(CAF_TARG(op, static_cast<int>)
<< ", " << CAF_ARG(fd) << ", " CAF_ARG(ptr)
CAF_LOG_TRACE(CAF_TSARG(op) << ", " << CAF_ARG(fd) << ", " CAF_ARG(ptr)
<< ", " << CAF_ARG(old_bf));
auto last = m_events.end();
auto i = std::lower_bound(m_events.begin(), last, fd, event_less{});
......@@ -384,6 +383,8 @@ class default_multiplexer : public multiplexer {
CAF_LOG_DEBUG("event has no effect (discarded): "
<< CAF_ARG(bf) << ", " << CAF_ARG(old_bf));
} else {
CAF_LOG_DEBUG("added handler " << ptr << " on fd " << fd << " for "
<< to_string(op) << " operations");
m_events.insert(i, event{fd, bf, ptr});
}
}
......@@ -534,7 +535,7 @@ class stream : public event_handler {
}
void stop_reading() {
CAF_LOGM_TRACE("caf::io::network::stream", "");
CAF_LOG_TRACE("");
m_sock.close_read();
backend().del(operation::read, m_sock.fd(), this);
}
......
......@@ -33,6 +33,12 @@ enum class operation {
propagate_error
};
constexpr const char* to_string(operation op) {
return op == operation::read ? "read"
: (op == operation::write ? "write"
: "propagate_error");
}
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -32,6 +32,12 @@ enum class protocol : uint32_t {
ipv6
};
constexpr const char* to_string(protocol value) {
return value == protocol::ethernet ? "ethernet"
: (value == protocol::ipv4 ? "ipv4"
: "ipv6");
}
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -82,15 +82,13 @@ behavior basp_broker::make_behavior() {
return {
// received from underlying broker implementation
[=](new_data_msg& msg) {
CAF_LOGM_TRACE("make_behavior$new_data_msg",
"handle = " << msg.handle.id());
CAF_LOG_TRACE("handle = " << msg.handle.id());
CAF_REQUIRE(m_ctx.count(msg.handle) > 0);
new_data(m_ctx[msg.handle], msg.buf);
},
// received from underlying broker implementation
[=](const new_connection_msg& msg) {
CAF_LOGM_TRACE("make_behavior$new_connection_msg",
"handle = " << msg.handle.id());
CAF_LOG_TRACE("handle = " << msg.handle.id());
CAF_REQUIRE(m_ctx.count(msg.handle) == 0);
auto& ctx = m_ctx[msg.handle];
ctx.hdl = msg.handle;
......@@ -100,8 +98,7 @@ behavior basp_broker::make_behavior() {
},
// received from underlying broker implementation
[=](const connection_closed_msg& msg) {
CAF_LOGM_TRACE("make_behavior$connection_closed_msg",
CAF_MARG(msg.handle, id));
CAF_LOG_TRACE(CAF_MARG(msg.handle, id));
auto j = m_ctx.find(msg.handle);
if (j != m_ctx.end()) {
auto hd = j->second.handshake_data;
......@@ -142,7 +139,7 @@ behavior basp_broker::make_behavior() {
},
// received from underlying broker implementation
[=](const acceptor_closed_msg& msg) {
CAF_LOGM_TRACE("make_behavior$acceptor_closed_msg", "");
CAF_LOG_TRACE("");
auto i = m_acceptors.find(msg.handle);
if (i == m_acceptors.end()) {
CAF_LOG_INFO("accept handle no longer in use");
......@@ -158,19 +155,16 @@ behavior basp_broker::make_behavior() {
const actor_addr& receiver,
message_id mid,
const message& msg) {
CAF_LOGM_TRACE("make_behavior$_Dispatch", "");
CAF_LOG_TRACE("");
dispatch(sender, receiver, mid, msg);
},
on(atom("_DelProxy"), arg_match) >> [=](const node_id& nid, actor_id aid) {
CAF_LOGM_TRACE("make_behavior$_DelProxy",
CAF_TSARG(nid) << ", "
<< CAF_ARG(aid));
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid));
erase_proxy(nid, aid);
},
// received from middleman actor
[=](put_atom, accept_handle hdl, const actor_addr& whom, uint16_t port) {
CAF_LOGM_TRACE("make_behavior$put_atom",
CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom)
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom)
<< ", " << CAF_ARG(port));
if (hdl.invalid() || whom == invalid_actor_addr) {
return;
......@@ -187,8 +181,7 @@ behavior basp_broker::make_behavior() {
},
[=](get_atom, connection_handle hdl, int64_t request_id,
actor client, std::set<std::string>& expected_ifs) {
CAF_LOGM_TRACE("make_behavior$get_atom",
CAF_ARG(hdl.id()) << ", " << CAF_ARG(request_id)
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", " << CAF_ARG(request_id)
<< ", " << CAF_TSARG(client));
try {
assign_tcp_scribe(hdl);
......@@ -211,8 +204,7 @@ behavior basp_broker::make_behavior() {
},
[=](delete_atom, int64_t request_id, const actor_addr& whom, uint16_t port)
-> message {
CAF_LOGM_TRACE("make_behavior$delete_atom",
CAF_ARG(request_id) << ", " << CAF_TSARG(whom)
CAF_LOG_TRACE(CAF_ARG(request_id) << ", " << CAF_TSARG(whom)
<< ", " << CAF_ARG(port));
if (whom == invalid_actor_addr) {
return make_message(error_atom::value, request_id,
......@@ -501,8 +493,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
auto mm = middleman::instance();
entry.first->attach_functor([=](uint32_t reason) {
mm->backend().dispatch([=] {
CAF_LOGM_TRACE("handle_basp_header$proxy_functor",
CAF_ARG(reason));
CAF_LOG_TRACE(CAF_ARG(reason));
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
bro->send_kill_proxy_instance(nid, aid, reason);
});
......@@ -713,8 +704,7 @@ void basp_broker::on_exit() {
}
void basp_broker::erase_proxy(const node_id& nid, actor_id aid) {
CAF_LOGM_TRACE("make_behavior$_DelProxy",
CAF_TSARG(nid) << ", " << CAF_ARG(aid));
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid));
m_namespace.erase(nid, aid);
if (m_namespace.empty()) {
CAF_LOG_DEBUG("no proxy left, request shutdown of connection");
......
......@@ -369,27 +369,34 @@ network::multiplexer& broker::backend() {
}
connection_handle broker::add_tcp_scribe(const std::string& hst, uint16_t prt) {
CAF_LOG_TRACE(CAF_ARG(hst) << ", " << CAF_ARG(prt));
return backend().add_tcp_scribe(this, hst, prt);
}
void broker::assign_tcp_scribe(connection_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_scribe(this, hdl);
}
connection_handle broker::add_tcp_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_scribe(this, fd);
}
void broker::assign_tcp_doorman(accept_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_doorman(this, hdl);
}
std::pair<accept_handle, uint16_t>
broker::add_tcp_doorman(uint16_t port, const char* in, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", in = " << (in ? in : "nullptr")
<< ", " << CAF_ARG(reuse_addr));
return backend().add_tcp_doorman(this, port, in, reuse_addr);
}
accept_handle broker::add_tcp_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_doorman(this, fd);
}
......
......@@ -413,14 +413,13 @@ namespace network {
std::vector<fd_event> poll_res;
while (!m_pollset.empty()) {
int presult;
CAF_LOG_DEBUG("poll() " << m_pollset.size() << " sockets");
# ifdef CAF_WINDOWS
presult = ::WSAPoll(m_pollset.data(), m_pollset.size(), -1);
# else
presult = ::poll(m_pollset.data(),
static_cast<nfds_t>(m_pollset.size()), -1);
# endif
CAF_LOG_DEBUG("poll() on " << m_pollset.size() << " sockets reported "
<< presult << " event(s)");
if (presult < 0) {
switch (last_socket_error()) {
case EINTR: {
......@@ -443,6 +442,7 @@ namespace network {
}
// scan pollset for events first, because we might alter m_pollset
// while running callbacks (not a good idea while traversing it)
CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < m_pollset.size() && presult > 0; ++i) {
auto& pfd = m_pollset[i];
if (pfd.revents != 0) {
......@@ -453,12 +453,14 @@ namespace network {
--presult; // stop as early as possible
}
}
CAF_LOG_DEBUG("found " << poll_res.size() << " IO event(s)");
for (auto& e : poll_res) {
// we try to read/write as much as possible by ignoring
// error states as long as there are still valid
// operations possible on the socket
handle_socket_event(e.fd, e.mask, e.ptr);
}
CAF_LOG_DEBUG("handle " << m_events.size() << " generated events");
poll_res.clear();
for (auto& me : m_events) {
handle(me);
......@@ -689,10 +691,6 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask,
del(operation::read, fd, nullptr);
}
}
CAF_LOG_DEBUG_IF(!checkerror && (mask & error_mask),
"ignored error because epoll still reported read or write "
"event; wait until no other event occurs before "
"handling error");
}
void default_multiplexer::init() {
......@@ -739,7 +737,7 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
m_stream.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_LOG_TRACE("");
m_stream.configure_read(config);
if (!m_launched) launch();
}
......@@ -750,16 +748,16 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
return m_stream.rd_buf();
}
void stop_reading() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_LOG_TRACE("");
m_stream.stop_reading();
disconnect(false);
}
void flush() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_LOG_TRACE("");
m_stream.flush(this);
}
void launch() {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_LOG_TRACE("");
CAF_REQUIRE(!m_launched);
m_launched = true;
m_stream.start(this);
......@@ -786,6 +784,7 @@ default_multiplexer::add_tcp_doorman(broker* self,
m_acceptor.init(std::move(s));
}
void new_connection() override {
CAF_LOG_TRACE("");
auto& dm = m_acceptor.backend();
accept_msg().handle
= dm.add_tcp_scribe(parent(), std::move(m_acceptor.accepted_socket()));
......@@ -793,10 +792,12 @@ default_multiplexer::add_tcp_doorman(broker* self,
m_accept_msg);
}
void stop_reading() override {
CAF_LOG_TRACE("");
m_acceptor.stop_reading();
disconnect(false);
}
void launch() override {
CAF_LOG_TRACE("");
m_acceptor.start(this);
}
private:
......@@ -813,19 +814,23 @@ connection_handle default_multiplexer::new_tcp_scribe(const std::string& host,
return connection_handle::from_int(int64_from_native_socket(fd));
}
void default_multiplexer::assign_tcp_scribe(broker* ptr,
void default_multiplexer::assign_tcp_scribe(broker* self,
connection_handle hdl) {
add_tcp_scribe(ptr, static_cast<native_socket>(hdl.id()));
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_MARG(hdl, id));
add_tcp_scribe(self, static_cast<native_socket>(hdl.id()));
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
return add_tcp_scribe(self, default_socket{*this, fd});
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host)
<< ", " << CAF_ARG(port));
return add_tcp_scribe(self, new_tcp_connection(host, port));
}
......@@ -1023,6 +1028,8 @@ in_port_t& port_of(sockaddr& what) {
template <int Family>
bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
CAF_LOGF_TRACE(CAF_ARG(fd) << ", " << CAF_ARG(host) << ", Family: "
<< (Family == AF_INET ? "AF_INET" : "AF_INET6"));
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
using sockaddr_type =
typename std::conditional<
......@@ -1040,7 +1047,8 @@ bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
optional<protocol> preferred) {
CAF_LOGF_TRACE(CAF_ARG(host) << ", " << CAF_ARG(port));
CAF_LOGF_TRACE(CAF_ARG(host) << ", " << CAF_ARG(port)
<< ", " << CAF_TSARG(preferred));
CAF_LOGF_INFO("try to connect to " << host << " on port " << port);
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
......@@ -1048,6 +1056,7 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
# endif
auto res = interfaces::native_address(host, preferred);
if (!res) {
CAF_LOGF_INFO("no such host");
throw network_error("no such host: " + host);
}
auto proto = res->second;
......@@ -1057,6 +1066,7 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
socket_guard sguard(fd);
if (proto == ipv6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
CAF_LOGF_INFO("successfully connected to host");
return fd;
}
// IPv4 fallback
......@@ -1162,7 +1172,8 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
std::pair<default_socket_acceptor, uint16_t>
new_tcp_acceptor(uint16_t port, const char* addr, bool reuse) {
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")
<< ", " << CAF_ARG(reuse));
auto& backend = get_multiplexer_singleton();
auto acceptor = new_tcp_acceptor_impl(port, addr, reuse);
auto bound_port = acceptor.second;
......
......@@ -33,6 +33,7 @@
#include <memory>
#include "caf/detail/logging.hpp"
#include "caf/detail/get_mac_addresses.hpp"
namespace caf {
......@@ -147,6 +148,7 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
optional<std::pair<std::string, protocol>>
interfaces::native_address(const std::string& host,
optional<protocol> preferred) {
CAF_LOGF_TRACE(CAF_ARG(host) << ", " << CAF_TSARG(preferred));
addrinfo hint;
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM;
......
......@@ -206,39 +206,18 @@ class middleman_actor_impl : public middleman_actor_base::base {
[=](delete_atom, const actor_addr& whom, uint16_t port) {
return del(whom, port);
},
[=](ok_atom ok, int64_t request_id) {
[=](ok_atom, int64_t request_id) {
// not legal for get results
CAF_REQUIRE(m_pending_gets.count(request_id) == 0);
auto i = m_pending_deletes.find(request_id);
if (i == m_pending_deletes.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(del_op_result{ok}.value);
m_pending_deletes.erase(i);
handle_ok<del_op_result>(m_pending_deletes, request_id);
},
[=](ok_atom ok, int64_t request_id, actor_addr result) {
[=](ok_atom, int64_t request_id, actor_addr& result) {
// not legal for delete results
CAF_REQUIRE(m_pending_deletes.count(request_id) == 0);
auto i = m_pending_gets.find(request_id);
if (i == m_pending_gets.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{ok, result}.value);
m_pending_gets.erase(i);
handle_ok<get_op_result>(m_pending_gets, request_id, std::move(result));
},
[=](error_atom error, int64_t request_id, std::string& reason) {
auto fget = [&](response_promise& rp) {
rp.deliver(get_op_result{error, std::move(reason)}.value);
};
auto fdel = [&](response_promise& rp) {
rp.deliver(del_op_result{error, std::move(reason)}.value);
};
if (!finalize_request(m_pending_gets, request_id, fget)
&& !finalize_request(m_pending_deletes, request_id, fdel)) {
CAF_LOG_ERROR("invalid request id: " << request_id);
}
[=](error_atom, int64_t request_id, std::string& reason) {
handle_error(request_id, reason);
}
};
}
......@@ -291,6 +270,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
}
del_op_promise del(const actor_addr& whom, uint16_t port = 0) {
CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port));
auto result = make_response_promise();
auto req_id = m_next_request_id++;
send(m_broker, delete_atom::value, req_id, whom, port);
......@@ -298,10 +278,24 @@ class middleman_actor_impl : public middleman_actor_base::base {
return result;
}
template <class T, class... Vs>
void handle_ok(map_type& storage, int64_t request_id, Vs&&... vs) {
CAF_LOG_TRACE(CAF_ARG(request_id));
auto i = storage.find(request_id);
if (i == storage.end()) {
CAF_LOG_ERROR("request id not found: " << request_id);
return;
}
i->second.deliver(T{ok_atom::value, std::forward<Vs>(vs)...}.value);
storage.erase(i);
}
template <class F>
bool finalize_request(map_type& storage, int64_t req_id, F fun) {
CAF_LOG_TRACE(CAF_ARG(req_id));
auto i = storage.find(req_id);
if (i == storage.end()) {
CAF_LOG_INFO("request ID not found in storage");
return false;
}
fun(i->second);
......@@ -309,6 +303,20 @@ class middleman_actor_impl : public middleman_actor_base::base {
return true;
}
void handle_error(int64_t request_id, std::string& reason) {
CAF_LOG_TRACE(CAF_ARG(request_id) << ", " << CAF_ARG(reason));
auto fget = [&](response_promise& rp) {
rp.deliver(get_op_result{error_atom::value, std::move(reason)}.value);
};
auto fdel = [&](response_promise& rp) {
rp.deliver(del_op_result{error_atom::value, std::move(reason)}.value);
};
if (!finalize_request(m_pending_gets, request_id, fget)
&& !finalize_request(m_pending_deletes, request_id, fdel)) {
CAF_LOG_ERROR("invalid request id: " << request_id);
}
}
actor m_broker;
middleman& m_parent;
int64_t m_next_request_id;
......@@ -338,7 +346,7 @@ void middleman::initialize() {
m_backend = network::multiplexer::make();
m_backend_supervisor = m_backend->make_supervisor();
m_thread = std::thread([this] {
CAF_LOGC_TRACE("caf::io::middleman", "initialize$run", "");
CAF_LOG_TRACE("");
m_backend->run();
});
m_backend->thread_id(m_thread.get_id());
......@@ -360,7 +368,7 @@ void middleman::initialize() {
void middleman::stop() {
CAF_LOG_TRACE("");
m_backend->dispatch([=] {
CAF_LOGC_TRACE("caf::io::middleman", "stop$lambda", "");
CAF_LOG_TRACE("");
// m_managers will be modified while we are stopping each manager,
// because each manager will call remove(...)
for (auto& kvp : m_named_brokers) {
......
......@@ -419,7 +419,9 @@ int main(int argc, char** argv) {
CAF_PRINTERR("usage: " << argv[0] << " [-s PORT|-c PORT1 PORT2 GROUP_PORT]");
}
});
CAF_CHECKPOINT();
await_all_actors_done();
CAF_CHECKPOINT();
shutdown();
// we either spawn a server or a client, in both cases
// there must have been exactly one dtor called
......
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