Commit 19abf110 authored by Marian Triebe's avatar Marian Triebe Committed by Dominik Charousset

Improve logging and fix cleanup during shutdown

Close #243
parent 8d753c01
......@@ -204,7 +204,7 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
#define CAF_PRINT4(arg0, arg1, arg2, arg3)
#else
#define CAF_PRINT4(lvlname, classname, funname, msg) \
caf::detail::logging::trace_helper { \
caf::detail::logging::trace_helper caf_log_trace_temporary_object { \
classname, funname, __FILE__, __LINE__, \
(caf::detail::oss_wr{} << msg).str() \
}
......
......@@ -227,10 +227,8 @@ void abstract_actor::cleanup(uint32_t reason) {
m_exit_reason = reason;
m_attachables_head.swap(head);
}
CAF_LOG_INFO_IF(!is_remote(), "cleanup actor with ID "
<< m_id << "; exit reason = "
<< reason << ", class = "
<< CAF_CLASS_NAME);
CAF_LOG_INFO_IF(!is_remote(), "cleanup actor with ID " << m_id
<< "; exit reason = " << reason);
// send exit messages
for (attachable* i = head.get(); i != nullptr; i = i->next.get()) {
i->actor_exited(this, reason);
......
......@@ -30,6 +30,13 @@
#include <sys/types.h>
#endif
#include "caf/config.hpp"
#if defined(CAF_LINUX) || defined(CAF_MACOS)
# include <cxxabi.h>
#endif
#include "caf/string_algorithms.hpp"
#include "caf/all.hpp"
......@@ -108,7 +115,33 @@ class logging_impl : public logging {
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) override {
# if defined(CAF_LINUX) || defined(CAF_MACOS)
int stat = 0;
std::unique_ptr<char, decltype(free)*> real_class_name{nullptr, free};
auto tmp = abi::__cxa_demangle(c_class_name, 0, 0, &stat);
real_class_name.reset(tmp);
std::string class_name = stat == 0 ? real_class_name.get() : c_class_name;
replace_all(class_name, " ", "");
replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymousnamespace)", "$anon$");
real_class_name.reset();
// hide CAF magic in logs
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, ',');
if (i != e) {
std::string substr(i + (prefix_end - prefix_begin), e);
class_name.swap(substr);
}
};
char prefix1[] = "caf.detail.proper_actor<";
char prefix2[] = "caf.detail.embedded<";
strip_magic(prefix1, prefix1 + (sizeof(prefix1) - 1));
strip_magic(prefix2, prefix2 + (sizeof(prefix2) - 1));
# else
std::string class_name = c_class_name;
# endif
replace_all(class_name, "::", ".");
replace_all(class_name, "(anonymous namespace)", "$anon$");
std::string file_name;
......
......@@ -81,7 +81,6 @@ void singletons::stop_singletons() {
stop(s_group_manager);
CAF_LOGF_DEBUG("stop type info map");
stop(s_uniform_type_info_map);
stop(s_logger);
stop(s_node_id);
// dispose singletons, i.e., release memory
dispose(s_scheduling_coordinator);
......@@ -91,6 +90,8 @@ void singletons::stop_singletons() {
dispose(s_actor_registry);
dispose(s_group_manager);
dispose(s_uniform_type_info_map);
// final steps
stop(s_logger);
dispose(s_logger);
dispose(s_node_id);
}
......
......@@ -75,7 +75,7 @@ basp_broker::basp_broker(middleman& pref) : broker(pref), m_namespace(*this) {
}
basp_broker::~basp_broker() {
// nop
CAF_LOG_TRACE("");
}
behavior basp_broker::make_behavior() {
......@@ -168,18 +168,32 @@ behavior basp_broker::make_behavior() {
erase_proxy(nid, aid);
},
// received from middleman actor
[=](put_atom, accept_handle hdl,
const actor_addr& whom, uint16_t port) {
[=](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_ARG(port));
if (hdl.invalid() || whom == invalid_actor_addr) {
return;
}
try {
assign_tcp_doorman(hdl);
}
catch (...) {
CAF_LOG_DEBUG("failed to assign doorman from handle");
return;
}
add_published_actor(hdl, actor_cast<abstract_actor_ptr>(whom), port);
parent().notify<hook::actor_published>(whom, port);
},
[=](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_TSARG(client));
try {
assign_tcp_scribe(hdl);
}
catch (std::exception&) {
catch (...) {
CAF_LOG_DEBUG("failed to assign scribe from handle");
send(client, error_atom::value, request_id,
"failed to assign scribe from handle");
......@@ -197,6 +211,9 @@ 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_ARG(port));
if (whom == invalid_actor_addr) {
return make_message(error_atom::value, request_id,
"whom == invalid_actor_addr");
......@@ -686,6 +703,7 @@ actor_proxy_ptr basp_broker::make_proxy(const node_id& nid, actor_id aid) {
}
void basp_broker::on_exit() {
CAF_LOG_TRACE("");
m_ctx.clear();
m_acceptors.clear();
m_open_ports.clear();
......
......@@ -36,7 +36,7 @@ namespace caf {
namespace io {
broker::servant::~servant() {
// nop
CAF_LOG_TRACE("");
}
broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
......
......@@ -582,24 +582,33 @@ void default_multiplexer::wr_dispatch_request(runnable* ptr) {
intptr_t ptrval = reinterpret_cast<intptr_t>(ptr);
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
::send(m_pipe.second, reinterpret_cast<socket_send_ptr>(&ptrval),
auto res = ::send(m_pipe.second, reinterpret_cast<socket_send_ptr>(&ptrval),
sizeof(ptrval), no_sigpipe_flag);
# else
auto unused = ::write(m_pipe.second, &ptrval, sizeof(ptrval));
static_cast<void>(unused);
auto res = ::write(m_pipe.second, &ptrval, sizeof(ptrval));
# endif
if (res <= 0) {
// pipe closed, discard runnable
ptr->request_deletion();
} else if (static_cast<size_t>(res) < sizeof(ptrval)) {
// must not happen: wrote invalid pointer to pipe
std::cerr << "[CAF] Fatal error: wrote invalid data to pipe" << std::endl;
abort();
}
}
default_multiplexer::runnable* default_multiplexer::rd_dispatch_request() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
::recv(m_pipe.first, reinterpret_cast<socket_recv_ptr>(&ptrval),
auto res = recv(m_pipe.first, reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto unused = ::read(m_pipe.first, &ptrval, sizeof(ptrval));
static_cast<void>(unused);
auto res = read(m_pipe.first, &ptrval, sizeof(ptrval));
# endif
if (res != sizeof(ptrval)) {
return nullptr;
}
return reinterpret_cast<runnable*>(ptrval);;
}
......@@ -699,8 +708,16 @@ default_multiplexer::~default_multiplexer() {
if (m_epollfd != invalid_native_socket) {
closesocket(m_epollfd);
}
closesocket(m_pipe.first);
// close write handle first
closesocket(m_pipe.second);
// flush pipe before closing it
nonblocking(m_pipe.first, true);
auto ptr = rd_dispatch_request();
while (ptr) {
ptr->request_deletion();
ptr = rd_dispatch_request();
}
closesocket(m_pipe.first);
# ifdef CAF_WINDOWS
WSACleanup();
# endif
......
......@@ -160,6 +160,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
~middleman_actor_impl();
void on_exit() {
CAF_LOG_TRACE("");
m_pending_gets.clear();
m_pending_deletes.clear();
m_broker = invalid_actor;
......@@ -246,6 +247,8 @@ class middleman_actor_impl : public middleman_actor_base::base {
either<ok_atom, uint16_t>::or_else<error_atom, std::string>
put(const actor_addr& whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port)
<< ", " << CAF_ARG(reuse_addr));
accept_handle hdl;
uint16_t actual_port;
try {
......@@ -269,6 +272,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
get_op_promise get(const std::string& hostname, uint16_t port,
std::set<std::string> expected_ifs) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
auto result = make_response_promise();
try {
auto hdl = m_parent.backend().new_tcp_scribe(hostname, port);
......@@ -313,10 +317,11 @@ class middleman_actor_impl : public middleman_actor_base::base {
};
middleman_actor_impl::~middleman_actor_impl() {
// nop
CAF_LOG_TRACE("");
}
middleman* middleman::instance() {
CAF_LOGF_TRACE("");
auto sid = detail::singletons::middleman_plugin_id;
auto fac = [] { return new middleman; };
auto res = detail::singletons::get_plugin_singleton(sid, fac);
......
......@@ -32,7 +32,7 @@ namespace io {
namespace network {
multiplexer::~multiplexer() {
// nop
CAF_LOG_TRACE("");
}
boost::asio::io_service* pimpl() {
......@@ -40,6 +40,7 @@ boost::asio::io_service* pimpl() {
}
multiplexer_ptr multiplexer::make() {
CAF_LOGF_TRACE("");
return multiplexer_ptr{new caf_multiplexer_impl};
}
......
......@@ -24,6 +24,7 @@
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
......@@ -34,7 +35,13 @@ namespace caf {
namespace io {
uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) {
const char* in, bool reuse) {
if (whom == nullptr) {
throw std::invalid_argument("cannot publish an invalid actor");
}
CAF_LOGF_TRACE("whom = " << to_string(whom->address())
<< ", " << CAF_ARG(port) << ", in = " << (in ? in : "")
<< ", " << CAF_ARG(reuse));
std::string str;
if (in != nullptr) {
str = in;
......@@ -42,7 +49,7 @@ uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
auto mm = get_middleman_actor();
scoped_actor self;
uint16_t result;
self->sync_send(mm, put_atom{}, whom->address(), port, str, reuse_addr).await(
self->sync_send(mm, put_atom::value, whom->address(), port, str, reuse).await(
[&](ok_atom, uint16_t res) {
result = res;
},
......
......@@ -29,17 +29,17 @@
namespace caf {
namespace io {
void unpublish_impl(const actor_addr& whom, uint16_t port, bool block_caller) {
void unpublish_impl(const actor_addr& whom, uint16_t port, bool blocking) {
CAF_LOGF_TRACE(CAF_TSARG(whom) << ", " << CAF_ARG(port) << CAF_ARG(blocking));
auto mm = get_middleman_actor();
if (block_caller) {
if (blocking) {
scoped_actor self;
self->sync_send(mm, delete_atom::value, whom, port).await(
[](ok_atom) {
// ok, basp_broker is done
},
[&](error_atom, const std::string& err) {
[](error_atom, const std::string&) {
// ok, basp_broker is done
CAF_LOGF_ERROR(err << ", " << CAF_ARG(port) << ", " << CAF_TSARG(whom));
}
);
} else {
......
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