Commit 5ac8118b authored by Dominik Charousset's avatar Dominik Charousset

fixed monitoring of remote actors

this patch ensures that local actors will receive exit and down
messages for remote actors correctly
parent e6df82dc
...@@ -325,3 +325,4 @@ unit_testing/test_tuple.cpp ...@@ -325,3 +325,4 @@ unit_testing/test_tuple.cpp
unit_testing/test_typed_spawn.cpp unit_testing/test_typed_spawn.cpp
unit_testing/test_uniform_type.cpp unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp unit_testing/test_yield_interface.cpp
cppa/io/event.hpp
...@@ -35,6 +35,8 @@ ...@@ -35,6 +35,8 @@
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
#include "cppa/io/event.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
/** /**
...@@ -102,9 +104,11 @@ class continuable { ...@@ -102,9 +104,11 @@ class continuable {
/** /**
* @brief Called from middleman before it removes this object * @brief Called from middleman before it removes this object
* due to an IO failure. * due to an IO failure, can be called twice
* (for read and for write error).
* @param bitmask Either @p read or @p write.
*/ */
virtual void io_failed() = 0; virtual void io_failed(event_bitmask bitmask) = 0;
protected: protected:
...@@ -122,7 +126,6 @@ class continuable { ...@@ -122,7 +126,6 @@ class continuable {
* inline and template member function implementations * * inline and template member function implementations *
******************************************************************************/ ******************************************************************************/
inline native_socket_type continuable::read_handle() const { inline native_socket_type continuable::read_handle() const {
return m_rd; return m_rd;
} }
......
...@@ -71,7 +71,7 @@ class default_peer : public extend<continuable>::with<buffered_writing> { ...@@ -71,7 +71,7 @@ class default_peer : public extend<continuable>::with<buffered_writing> {
void dispose() override; void dispose() override;
void io_failed() override; void io_failed(event_bitmask mask) override;
void enqueue(const message_header& hdr, const any_tuple& msg); void enqueue(const message_header& hdr, const any_tuple& msg);
......
...@@ -56,7 +56,7 @@ class default_peer_acceptor : public continuable { ...@@ -56,7 +56,7 @@ class default_peer_acceptor : public continuable {
void dispose() override; void dispose() override;
void io_failed() override; void io_failed(event_bitmask) override;
private: private:
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 EVENT_HPP
#define EVENT_HPP
namespace cppa { namespace io {
typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace <anonymous>::event
template<unsigned InputEvent, unsigned OutputEvent, unsigned ErrorEvent>
inline event_bitmask from_int_bitmask(unsigned mask) {
event_bitmask result = event::none;
// read/write as long as possible
if (mask & InputEvent) result = event::read;
if (mask & OutputEvent) result |= event::write;
if (result == event::none && mask & ErrorEvent) result = event::error;
return result;
}
} } // namespace cppa::io
#endif // EVENT_HPP
...@@ -37,32 +37,11 @@ ...@@ -37,32 +37,11 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/io/event.hpp"
#include "cppa/io/continuable.hpp" #include "cppa/io/continuable.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
typedef int event_bitmask;
namespace event { namespace {
constexpr event_bitmask none = 0x00;
constexpr event_bitmask read = 0x01;
constexpr event_bitmask write = 0x02;
constexpr event_bitmask both = 0x03;
constexpr event_bitmask error = 0x04;
} } // namespace event
template<unsigned InputEvent, unsigned OutputEvent, unsigned ErrorEvent>
inline event_bitmask from_int_bitmask(unsigned mask) {
event_bitmask result = event::none;
// read/write as long as possible
if (mask & InputEvent) result = event::read;
if (mask & OutputEvent) result |= event::write;
if (result == event::none && mask & ErrorEvent) result = event::error;
return result;
}
struct fd_meta_info { struct fd_meta_info {
native_socket_type fd; native_socket_type fd;
continuable* ptr; continuable* ptr;
......
...@@ -69,6 +69,9 @@ struct optional_variant_copy_helper { ...@@ -69,6 +69,9 @@ struct optional_variant_copy_helper {
inline void operator()() const { inline void operator()() const {
lhs = unit; lhs = unit;
} }
inline void operator()(const none_t&) const {
lhs = none;
}
}; };
template<typename T> template<typename T>
......
...@@ -193,6 +193,7 @@ void actor::cleanup(std::uint32_t reason) { ...@@ -193,6 +193,7 @@ void actor::cleanup(std::uint32_t reason) {
<< ", class = " << detail::demangle(typeid(*this))); << ", class = " << detail::demangle(typeid(*this)));
// send exit messages // send exit messages
auto msg = make_any_tuple(atom("EXIT"), reason); auto msg = make_any_tuple(atom("EXIT"), reason);
CPPA_LOGM_DEBUG("cppa::actor", "send EXIT to " << mlinks.size() << " links");
for (actor_ptr& aptr : mlinks) { for (actor_ptr& aptr : mlinks) {
message_header hdr{this, aptr, message_priority::high}; message_header hdr{this, aptr, message_priority::high};
hdr.deliver(msg); hdr.deliver(msg);
......
...@@ -113,8 +113,8 @@ class broker::servant : public continuable { ...@@ -113,8 +113,8 @@ class broker::servant : public continuable {
: super{std::forward<Ts>(args)...}, m_disconnected{false} : super{std::forward<Ts>(args)...}, m_disconnected{false}
, m_parent{move(parent)} { } , m_parent{move(parent)} { }
void io_failed() override { void io_failed(event_bitmask mask) override {
disconnect(); if (mask == event::read) disconnect();
} }
void dispose() override { void dispose() override {
......
...@@ -77,9 +77,11 @@ default_peer::default_peer(default_protocol* parent, ...@@ -77,9 +77,11 @@ default_peer::default_peer(default_protocol* parent,
m_meta_msg = uniform_typeid<any_tuple>(); m_meta_msg = uniform_typeid<any_tuple>();
} }
void default_peer::io_failed() { void default_peer::io_failed(event_bitmask mask) {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr")); CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr")
if (m_node) { << " mask = " << mask);
// make sure this code is executed only once by filtering for read failure
if (mask == event::read && m_node) {
// kill all proxies // kill all proxies
auto& children = m_parent->addressing()->proxies(*m_node); auto& children = m_parent->addressing()->proxies(*m_node);
for (auto& kvp : children) { for (auto& kvp : children) {
......
...@@ -83,7 +83,7 @@ continue_reading_result default_peer_acceptor::continue_reading() { ...@@ -83,7 +83,7 @@ continue_reading_result default_peer_acceptor::continue_reading() {
} }
} }
void default_peer_acceptor::io_failed() { void default_peer_acceptor::io_failed(event_bitmask) {
CPPA_LOG_INFO("removed default_peer_acceptor " CPPA_LOG_INFO("removed default_peer_acceptor "
<< this << " due to an IO failure"); << this << " due to an IO failure");
} }
......
...@@ -232,7 +232,9 @@ class middleman_overseer : public continuable { ...@@ -232,7 +232,9 @@ class middleman_overseer : public continuable {
return read_continue_later; return read_continue_later;
} }
void io_failed() { CPPA_CRITICAL("IO on pipe failed"); } void io_failed(event_bitmask) override {
CPPA_CRITICAL("IO on pipe failed");
}
private: private:
...@@ -305,8 +307,10 @@ void middleman_loop(middleman_impl* impl) { ...@@ -305,8 +307,10 @@ void middleman_loop(middleman_impl* impl) {
case event::write: { case event::write: {
CPPA_LOGF_DEBUG("handle event::write for " << io); CPPA_LOGF_DEBUG("handle event::write for " << io);
switch (io->continue_writing()) { switch (io->continue_writing()) {
case read_failure:
io->io_failed(event::write);
// fall through
case write_closed: case write_closed:
case write_failure:
impl->stop_writer(io); impl->stop_writer(io);
CPPA_LOGF_DEBUG("writer removed because of error"); CPPA_LOGF_DEBUG("writer removed because of error");
break; break;
...@@ -322,8 +326,10 @@ void middleman_loop(middleman_impl* impl) { ...@@ -322,8 +326,10 @@ void middleman_loop(middleman_impl* impl) {
case event::read: { case event::read: {
CPPA_LOGF_DEBUG("handle event::read for " << io); CPPA_LOGF_DEBUG("handle event::read for " << io);
switch (io->continue_reading()) { switch (io->continue_reading()) {
case read_closed:
case read_failure: case read_failure:
io->io_failed(event::read);
// fall through
case read_closed:
impl->stop_reader(io); impl->stop_reader(io);
CPPA_LOGF_DEBUG("remove peer"); CPPA_LOGF_DEBUG("remove peer");
break; break;
...@@ -333,7 +339,8 @@ void middleman_loop(middleman_impl* impl) { ...@@ -333,7 +339,8 @@ void middleman_loop(middleman_impl* impl) {
} }
case event::error: { case event::error: {
CPPA_LOGF_DEBUG("event::error; remove peer " << io); CPPA_LOGF_DEBUG("event::error; remove peer " << io);
io->io_failed(); io->io_failed(event::write);
io->io_failed(event::read);
impl->stop_reader(io); impl->stop_reader(io);
impl->stop_writer(io); impl->stop_writer(io);
} }
...@@ -354,8 +361,10 @@ void middleman_loop(middleman_impl* impl) { ...@@ -354,8 +361,10 @@ void middleman_loop(middleman_impl* impl) {
switch (mask) { switch (mask) {
case event::write: case event::write:
switch (io->continue_writing()) { switch (io->continue_writing()) {
case write_closed:
case write_failure: case write_failure:
io->io_failed(event::write);
// fall through
case write_closed:
case write_done: case write_done:
handler->erase_later(io, event::write); handler->erase_later(io, event::write);
break; break;
...@@ -363,7 +372,8 @@ void middleman_loop(middleman_impl* impl) { ...@@ -363,7 +372,8 @@ void middleman_loop(middleman_impl* impl) {
} }
break; break;
case event::error: case event::error:
io->io_failed(); io->io_failed(event::write);
io->io_failed(event::read);
handler->erase_later(io, event::both); handler->erase_later(io, event::both);
break; break;
default: default:
......
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