Commit ff7cf3c4 authored by Dominik Charousset's avatar Dominik Charousset

fixed issues in middleman regarding io failures

this patch fixes an issue in the middleman where it didn't report back
on socket IO failures, why actors never received 'EXIT' messages in case
of disconnect events; this patch also replaces continuable_writer with
continuable_io, which derives continuable_reader now to avoid the
pointless multiple inheritance
parent 6d860687
...@@ -282,7 +282,7 @@ cppa/network/default_protocol.hpp ...@@ -282,7 +282,7 @@ cppa/network/default_protocol.hpp
src/default_protocol.cpp src/default_protocol.cpp
cppa/logging.hpp cppa/logging.hpp
src/logging.cpp src/logging.cpp
cppa/network/continuable_writer.hpp cppa/network/continuable_io.hpp
src/continuable_writer.cpp src/continuable_writer.cpp
cppa/network/middleman_event_handler_base.hpp cppa/network/middleman_event_handler_base.hpp
src/actor_addressing.cpp src/actor_addressing.cpp
......
...@@ -34,12 +34,13 @@ ...@@ -34,12 +34,13 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
#include "cppa/network/continuable_reader.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
/** /**
* @brief Denotes the return value of * @brief Denotes the return value of
* {@link continuable_writer::continue_writing()}. * {@link continuable_io::continue_writing()}.
*/ */
enum continue_writing_result { enum continue_writing_result {
write_failure, write_failure,
...@@ -49,11 +50,11 @@ enum continue_writing_result { ...@@ -49,11 +50,11 @@ enum continue_writing_result {
}; };
/** /**
* @brief An object performing asynchronous output on a file handle. * @brief An object performing asynchronous input and output.
*/ */
class continuable_writer : virtual public ref_counted { class continuable_io : public continuable_reader {
typedef ref_counted super; typedef continuable_reader super;
public: public:
...@@ -64,6 +65,8 @@ class continuable_writer : virtual public ref_counted { ...@@ -64,6 +65,8 @@ class continuable_writer : virtual public ref_counted {
return m_wr; return m_wr;
} }
continuable_io* as_io();
/** /**
* @brief Writes to {@link write_handle()}. * @brief Writes to {@link write_handle()}.
*/ */
...@@ -71,7 +74,7 @@ class continuable_writer : virtual public ref_counted { ...@@ -71,7 +74,7 @@ class continuable_writer : virtual public ref_counted {
protected: protected:
continuable_writer(native_socket_type write_handle); continuable_io(native_socket_type read_fd, native_socket_type write_fd);
private: private:
...@@ -79,7 +82,7 @@ class continuable_writer : virtual public ref_counted { ...@@ -79,7 +82,7 @@ class continuable_writer : virtual public ref_counted {
}; };
typedef intrusive_ptr<continuable_writer> continuable_writer_ptr; typedef intrusive_ptr<continuable_io> continuable_io_ptr;
} } // namespace cppa::network } } // namespace cppa::network
......
...@@ -52,12 +52,12 @@ enum continue_reading_result { ...@@ -52,12 +52,12 @@ enum continue_reading_result {
read_continue_later read_continue_later
}; };
class continuable_writer; class continuable_io;
/** /**
* @brief An object performing asynchronous input on a file handle. * @brief An object performing asynchronous input.
*/ */
class continuable_reader : virtual public ref_counted { class continuable_reader : public ref_counted {
public: public:
...@@ -72,14 +72,20 @@ class continuable_reader : virtual public ref_counted { ...@@ -72,14 +72,20 @@ class continuable_reader : virtual public ref_counted {
virtual continue_reading_result continue_reading() = 0; virtual continue_reading_result continue_reading() = 0;
/** /**
* @return Casts @p this to a continuable_writer, returns @p nullptr * @brief Casts @p this to a continuable_io or returns @p nullptr
* if cast fails. * if cast fails.
*/ */
virtual continuable_writer* as_writer(); virtual continuable_io* as_io();
/**
* @brief Called from middleman before it removes this object
* due to an IO failure.
*/
virtual void io_failed() = 0;
protected: protected:
continuable_reader(native_socket_type rd); continuable_reader(native_socket_type read_fd);
private: private:
......
...@@ -44,17 +44,16 @@ ...@@ -44,17 +44,16 @@
#include "cppa/network/input_stream.hpp" #include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp" #include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp" #include "cppa/network/continuable_io.hpp"
#include "cppa/network/default_message_queue.hpp" #include "cppa/network/default_message_queue.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
class default_protocol; class default_protocol;
class default_peer : public continuable_reader, public continuable_writer { class default_peer : public continuable_io {
typedef continuable_reader lsuper; typedef continuable_io super;
typedef continuable_writer rsuper;
friend class default_protocol; friend class default_protocol;
...@@ -69,7 +68,9 @@ class default_peer : public continuable_reader, public continuable_writer { ...@@ -69,7 +68,9 @@ class default_peer : public continuable_reader, public continuable_writer {
continue_writing_result continue_writing(); continue_writing_result continue_writing();
continuable_writer* as_writer(); continuable_io* as_io();
void io_failed();
void enqueue(const message_header& hdr, const any_tuple& msg); void enqueue(const message_header& hdr, const any_tuple& msg);
......
...@@ -54,6 +54,8 @@ class default_peer_acceptor : public continuable_reader { ...@@ -54,6 +54,8 @@ class default_peer_acceptor : public continuable_reader {
inline const actor_ptr& published_actor() const { return m_pa; } inline const actor_ptr& published_actor() const { return m_pa; }
void io_failed();
private: private:
default_protocol* m_parent; default_protocol* m_parent;
......
...@@ -39,7 +39,7 @@ ...@@ -39,7 +39,7 @@
#include "cppa/network/protocol.hpp" #include "cppa/network/protocol.hpp"
#include "cppa/network/acceptor.hpp" #include "cppa/network/acceptor.hpp"
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace detail { class singleton_manager; } } namespace cppa { namespace detail { class singleton_manager; } }
......
...@@ -38,7 +38,7 @@ ...@@ -38,7 +38,7 @@
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
...@@ -104,17 +104,17 @@ class middleman_event_handler_base { ...@@ -104,17 +104,17 @@ class middleman_event_handler_base {
fd = ptr->read_handle(); fd = ptr->read_handle();
break; break;
case event::write: { case event::write: {
auto wptr = ptr->as_writer(); auto wptr = ptr->as_io();
if (wptr) fd = wptr->write_handle(); if (wptr) fd = wptr->write_handle();
else { else {
CPPA_LOG_ERROR("ptr->as_writer() returned nullptr"); CPPA_LOG_ERROR("ptr->downcast() returned nullptr");
return; return;
} }
break; break;
} }
case event::both: { case event::both: {
fd = ptr->read_handle(); fd = ptr->read_handle();
auto wptr = ptr->as_writer(); auto wptr = ptr->as_io();
if (wptr) { if (wptr) {
auto wrfd = wptr->write_handle(); auto wrfd = wptr->write_handle();
if (fd != wrfd) { if (fd != wrfd) {
...@@ -126,7 +126,7 @@ class middleman_event_handler_base { ...@@ -126,7 +126,7 @@ class middleman_event_handler_base {
} }
} }
else { else {
CPPA_LOG_ERROR("ptr->as_writer() returned nullptr"); CPPA_LOG_ERROR("ptr->downcast() returned nullptr");
return; return;
} }
break; break;
...@@ -229,12 +229,16 @@ class event_iterator_impl { ...@@ -229,12 +229,16 @@ class event_iterator_impl {
return m_access.type(m_i); return m_access.type(m_i);
} }
inline void io_failed() {
ptr()->io_failed();
}
inline continue_reading_result continue_reading() { inline continue_reading_result continue_reading() {
return ptr()->continue_reading(); return ptr()->continue_reading();
} }
inline continue_writing_result continue_writing() { inline continue_writing_result continue_writing() {
return ptr()->as_writer()->continue_writing(); return ptr()->as_io()->continue_writing();
} }
inline bool equal_to(const event_iterator_impl& other) const { inline bool equal_to(const event_iterator_impl& other) const {
......
...@@ -48,7 +48,7 @@ namespace cppa { namespace network { ...@@ -48,7 +48,7 @@ namespace cppa { namespace network {
class abstract_middleman; class abstract_middleman;
class continuable_reader; class continuable_reader;
class continuable_writer; class continuable_io;
/** /**
* @brief Implements a communication protocol. * @brief Implements a communication protocol.
......
...@@ -29,13 +29,13 @@ ...@@ -29,13 +29,13 @@
#include "cppa/network/continuable_reader.hpp" #include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_writer.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
continuable_reader::continuable_reader(native_socket_type rd) : m_rd(rd) { } continuable_reader::continuable_reader(native_socket_type rd) : m_rd(rd) { }
continuable_writer* continuable_reader::as_writer() { return nullptr; } continuable_io* continuable_reader::as_io() { return nullptr; }
} } // namespace cppa::network } } // namespace cppa::network
...@@ -28,10 +28,13 @@ ...@@ -28,10 +28,13 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/network/continuable_writer.hpp" #include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network { namespace cppa { namespace network {
continuable_writer::continuable_writer(native_socket_type wr) : m_wr(wr) { } continuable_io::continuable_io(native_socket_type rd, native_socket_type wr)
: super(rd), m_wr(wr) { }
continuable_io* continuable_io::as_io() { return this; }
} } // namespace cppa::network } } // namespace cppa::network
...@@ -58,7 +58,7 @@ default_peer::default_peer(default_protocol* parent, ...@@ -58,7 +58,7 @@ default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in, const input_stream_ptr& in,
const output_stream_ptr& out, const output_stream_ptr& out,
process_information_ptr peer_ptr) process_information_ptr peer_ptr)
: lsuper(in->read_handle()), rsuper(out->write_handle()) : super(in->read_handle(), out->write_handle())
, m_parent(parent), m_in(in), m_out(out) , m_parent(parent), m_in(in), m_out(out)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info) , m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_node(peer_ptr) , m_node(peer_ptr)
...@@ -92,6 +92,11 @@ void default_peer::disconnected() { ...@@ -92,6 +92,11 @@ void default_peer::disconnected() {
} }
} }
void default_peer::io_failed() {
CPPA_LOG_TRACE("");
disconnected();
}
continue_reading_result default_peer::continue_reading() { continue_reading_result default_peer::continue_reading() {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
for (;;) { for (;;) {
...@@ -277,31 +282,39 @@ void default_peer::deliver(const message_header& hdr, any_tuple msg) { ...@@ -277,31 +282,39 @@ void default_peer::deliver(const message_header& hdr, any_tuple msg) {
void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) { void default_peer::link(const actor_ptr& sender, const actor_ptr& ptr) {
// this message is sent from default_actor_proxy in link_to and
// establish_backling to cause the original actor (sender) to establish
// a link to ptr as well
CPPA_LOG_TRACE(CPPA_MARG(sender, get) CPPA_LOG_TRACE(CPPA_MARG(sender, get)
<< ", " << CPPA_MARG(ptr, get)); << ", " << CPPA_MARG(ptr, get));
CPPA_LOG_ERROR_IF(!sender, "received 'LINK' from invalid sender");
CPPA_LOG_ERROR_IF(!ptr, "received 'LINK' with invalid target");
if (!sender || !ptr) return;
CPPA_LOG_ERROR_IF(sender->is_proxy(),
"received 'LINK' for a non-local actor");
if (ptr->is_proxy()) {
// make sure to not send a needless 'LINK' message back
ptr.downcast<actor_proxy>()->local_link_to(sender);
}
else sender->link_to(ptr);
if (ptr && sender && sender->is_proxy()) { if (ptr && sender && sender->is_proxy()) {
sender.downcast<actor_proxy>()->local_link_to(ptr); sender.downcast<actor_proxy>()->local_link_to(ptr);
} }
else {
CPPA_LOG_ERROR_IF(!sender,
"received LINK from invalid sender");
CPPA_LOG_ERROR_IF(sender && !sender->is_proxy(),
"sender is not a proxy");
CPPA_LOG_ERROR_IF(!ptr, "received LINK with invalid target");
}
} }
void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) { void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
if (ptr && sender && sender->is_proxy()) { CPPA_LOG_TRACE(CPPA_MARG(sender, get)
sender.downcast<actor_proxy>()->local_unlink_from(ptr); << ", " << CPPA_MARG(ptr, get));
} CPPA_LOG_ERROR_IF(!sender, "received 'UNLINK' from invalid sender");
else { CPPA_LOG_ERROR_IF(!ptr, "received 'UNLINK' with invalid target");
CPPA_LOG_ERROR_IF(!sender, if (!sender || !ptr) return;
"received UNLINK from invalid sender"); CPPA_LOG_ERROR_IF(sender->is_proxy(),
CPPA_LOG_ERROR_IF(sender && !sender->is_proxy(), "received 'UNLINK' for a non-local actor");
"received UNLINK but sender is not a proxy"); if (ptr->is_proxy()) {
CPPA_LOG_ERROR_IF(!ptr, "received UNLINK with invalid target"); // make sure to not send a needles 'UNLINK' message back
ptr.downcast<actor_proxy>()->local_unlink_from(sender);
} }
else sender->unlink_from(ptr);
} }
continue_writing_result default_peer::continue_writing() { continue_writing_result default_peer::continue_writing() {
...@@ -341,7 +354,7 @@ continue_writing_result default_peer::continue_writing() { ...@@ -341,7 +354,7 @@ continue_writing_result default_peer::continue_writing() {
return write_done; return write_done;
} }
continuable_writer* default_peer::as_writer() { continuable_io* default_peer::as_io() {
return this; return this;
} }
......
...@@ -46,8 +46,8 @@ using namespace std; ...@@ -46,8 +46,8 @@ using namespace std;
namespace cppa { namespace network { namespace cppa { namespace network {
default_peer_acceptor::default_peer_acceptor(default_protocol* parent, default_peer_acceptor::default_peer_acceptor(default_protocol* parent,
acceptor_uptr aur, acceptor_uptr aur,
const actor_ptr& pa) const actor_ptr& pa)
: super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur)), m_pa(pa) { } : super(aur->file_handle()), m_parent(parent), m_ptr(std::move(aur)), m_pa(pa) { }
continue_reading_result default_peer_acceptor::continue_reading() { continue_reading_result default_peer_acceptor::continue_reading() {
...@@ -83,4 +83,9 @@ continue_reading_result default_peer_acceptor::continue_reading() { ...@@ -83,4 +83,9 @@ continue_reading_result default_peer_acceptor::continue_reading() {
} }
} }
void default_peer_acceptor::io_failed() {
CPPA_LOG_INFO("removed default_peer_acceptor "
<< this << " due to an IO failure");
}
} } // namespace cppa::network } } // namespace cppa::network
...@@ -515,6 +515,8 @@ class middleman_overseer : public continuable_reader { ...@@ -515,6 +515,8 @@ class middleman_overseer : public continuable_reader {
return read_continue_later; return read_continue_later;
} }
void io_failed() { CPPA_CRITICAL("IO on pipe failed"); }
private: private:
middleman_queue& m_queue; middleman_queue& m_queue;
...@@ -529,13 +531,13 @@ middleman_event_handler& abstract_middleman::handler() { ...@@ -529,13 +531,13 @@ middleman_event_handler& abstract_middleman::handler() {
void abstract_middleman::continue_writer(const continuable_reader_ptr& ptr) { void abstract_middleman::continue_writer(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get()); CPPA_LOG_TRACE("ptr = " << ptr.get());
CPPA_REQUIRE(ptr->as_writer() != nullptr); CPPA_REQUIRE(ptr->as_io() != nullptr);
handler().add(ptr, event::write); handler().add(ptr, event::write);
} }
void abstract_middleman::stop_writer(const continuable_reader_ptr& ptr) { void abstract_middleman::stop_writer(const continuable_reader_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get()); CPPA_LOG_TRACE("ptr = " << ptr.get());
CPPA_REQUIRE(ptr->as_writer() != nullptr); CPPA_REQUIRE(ptr->as_io() != nullptr);
handler().erase(ptr, event::write); handler().erase(ptr, event::write);
} }
...@@ -603,6 +605,7 @@ void middleman_loop(middleman_impl* impl) { ...@@ -603,6 +605,7 @@ void middleman_loop(middleman_impl* impl) {
} }
case event::error: { case event::error: {
CPPA_LOGF_DEBUG("event::error; remove peer " << i->ptr()); CPPA_LOGF_DEBUG("event::error; remove peer " << i->ptr());
i->io_failed();
impl->stop_reader(i->ptr()); impl->stop_reader(i->ptr());
impl->stop_writer(i->ptr()); impl->stop_writer(i->ptr());
} }
...@@ -633,6 +636,7 @@ void middleman_loop(middleman_impl* impl) { ...@@ -633,6 +636,7 @@ void middleman_loop(middleman_impl* impl) {
} }
break; break;
case event::error: case event::error:
i->io_failed();
handler->erase(i->ptr(), event::both); handler->erase(i->ptr(), event::both);
break; break;
default: default:
......
...@@ -47,7 +47,7 @@ void protocol::continue_reader(continuable_reader* ptr) { ...@@ -47,7 +47,7 @@ void protocol::continue_reader(continuable_reader* ptr) {
void protocol::continue_writer(continuable_reader* ptr) { void protocol::continue_writer(continuable_reader* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr)); CPPA_LOG_TRACE(CPPA_ARG(ptr));
CPPA_REQUIRE(ptr->as_writer() != nullptr); CPPA_REQUIRE(ptr->as_io() != nullptr);
m_parent->continue_writer(ptr); m_parent->continue_writer(ptr);
} }
......
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