Commit 60429ff3 authored by Dominik Charousset's avatar Dominik Charousset

improved logging of broker-related functions

parent 2ab72626
......@@ -249,9 +249,11 @@ class broker::doorman : public broker::servant {
doorman(broker_ptr parent, acceptor_uptr ptr)
: super{move(parent), ptr->file_handle()}
, m_ptr{move(ptr)}
//, m_ptr{move(ptr)}
, m_accept_msg{atom("IO_accept"),
accept_handle::from_int(m_ptr->file_handle())} { }
accept_handle::from_int(ptr->file_handle())} {
m_ptr.reset(ptr.release());
}
continue_reading_result continue_reading() override {
CPPA_LOG_TRACE("");
......@@ -404,12 +406,20 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
scoped_self_setter sss{ptr.get()};
ptr->init();
// continue reader only if not inherited from default_broker_impl
CPPA_LOGF_WARNING_IF(!ptr->has_behavior(), "broker w/o behavior spawned");
auto mm = get_middleman();
if (ptr->has_behavior()) {
mm->run_later([=] {
CPPA_LOGC_TRACE("NONE", "init_and_launch::run_later_functor", "");
CPPA_LOGF_WARNING_IF(ptr->m_io.empty() && ptr->m_accept.empty(),
"both m_io and m_accept are empty");
// 'launch' all backends
CPPA_LOGC_DEBUG("NONE", "init_and_launch::run_later_functor",
"add " << ptr->m_io.size() << " IO servants");
for (auto& kvp : ptr->m_io)
mm->continue_reader(kvp.second.get());
CPPA_LOGC_DEBUG("NONE", "init_and_launch::run_later_functor",
"add " << ptr->m_accept.size() << " acceptors");
for (auto& kvp : ptr->m_accept)
mm->continue_reader(kvp.second.get());
});
......
......@@ -32,6 +32,8 @@
#include <cstring>
#include <errno.h>
#include <iostream>
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
#include "cppa/io/stream.hpp"
......@@ -104,6 +106,8 @@ ipv4_acceptor::ipv4_acceptor(native_socket_type fd, bool nonblocking)
std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port,
const char* addr) {
CPPA_LOGM_TRACE("ipv4_acceptor", CPPA_ARG(port) << ", addr = "
<< (addr ? addr : "nullptr"));
native_socket_type sockfd;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) {
......@@ -136,6 +140,7 @@ std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port,
nonblocking(sockfd, true);
// ok, no exceptions
sguard.release();
CPPA_LOGM_DEBUG("ipv4_acceptor", "sockfd = " << sockfd);
return std::unique_ptr<ipv4_acceptor>(new ipv4_acceptor(sockfd, true));
}
......
......@@ -122,7 +122,8 @@ io::stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
}
io::stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) {
std::uint16_t port) {
CPPA_LOGF_TRACE(CPPA_ARG(host) << ", " << CPPA_ARG(port));
CPPA_LOGF_INFO("try to connect to " << host << " on port " << port);
struct sockaddr_in serv_addr;
struct hostent* server;
......@@ -140,9 +141,13 @@ io::stream_ptr ipv4_io_stream::connect_to(const char* host,
serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
serv_addr.sin_port = htons(port);
CPPA_LOGF_DEBUG("call connect()");
if (connect(fd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0) {
CPPA_LOGF_ERROR("could not connect to to " << host
<< " on port " << port);
throw network_error("could not connect to host");
}
CPPA_LOGF_DEBUG("enable nodelay + nonblocking for socket");
tcp_nodelay(fd, true);
nonblocking(fd, true);
return new ipv4_io_stream(fd);
......
......@@ -220,14 +220,13 @@ class middleman_overseer : public continuable {
CPPA_CRITICAL("cannot read from pipe");
}
}
atomic_thread_fence(memory_order_seq_cst);
for (int i = 0; i < read_result; ++i) {
unique_ptr<middleman_event> msg(m_queue.try_pop());
if (!msg) {
CPPA_LOGMF(CPPA_ERROR, self, "nullptr dequeued");
CPPA_CRITICAL("nullptr dequeued");
}
CPPA_LOGMF(CPPA_DEBUG, self, "execute run_later functor");
CPPA_LOGF_DEBUG("execute run_later functor");
(*msg)();
}
return read_continue_later;
......
......@@ -79,7 +79,9 @@ void middleman_event_handler::alteration(continuable* ptr,
}
void middleman_event_handler::add_later(continuable* ptr, event_bitmask e) {
CPPA_LOG_TRACE(CPPA_ARG(ptr) << ", " << CPPA_TARG(e, eb2str));
CPPA_LOG_TRACE(CPPA_ARG(ptr) << ", "
<< CPPA_TARG(e, eb2str)
<< ", socket = " << ptr->read_handle());
alteration(ptr, e, fd_meta_event::add);
}
......
......@@ -39,6 +39,11 @@ using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void report_unexpected() {
std::cerr << "unexpected message: " << to_string(self->last_dequeued())
<< std::endl;
}
void ping(size_t num_pings) {
auto count = std::make_shared<size_t>(0);
become (
......@@ -48,9 +53,11 @@ void ping(size_t num_pings) {
on(atom("pong"), arg_match) >> [=](int value) {
if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1);
}
},
others() >> report_unexpected
);
}
},
others() >> report_unexpected
);
}
......@@ -65,9 +72,11 @@ void pong() {
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
self->quit(reason);
}
},
others() >> report_unexpected
);
}
},
others() >> report_unexpected
);
}
......@@ -106,21 +115,18 @@ void peer(io::broker* thisptr, const actor_ptr& buddy) {
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (thisptr->last_sender() == buddy) self->quit(reason);
},
others() >> [] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
others() >> report_unexpected
);
}
void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) {
CPPA_LOGF_INFO("received IO_accept");
thisptr->fork(peer, hdl, buddy);
self->quit();
},
others() >> [] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
others() >> report_unexpected
);
}
......
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