Commit f0a76465 authored by Dominik Charousset's avatar Dominik Charousset

Make sure brokers stop their backends on exit

This patch makes sure that a broker stops all of its backend when caling
`self->quit()`. Furthermore, this patch uses `cppa::network_error` instead of
`std::ios_base::failure` and adds a new exception `stream_at_eof` to get better
diagnostic in logfiles.
parent ae3395da
...@@ -32,22 +32,22 @@ namespace fd_util { ...@@ -32,22 +32,22 @@ namespace fd_util {
std::string last_socket_error_as_string(); std::string last_socket_error_as_string();
// throws ios_base::failure and adds errno failure if @p add_errno_failure // throws network_error and adds errno failure if @p add_errno_failure
void throw_io_failure [[noreturn]] (const char* what, bool add_errno = true); void throw_io_failure [[noreturn]] (const char* what, bool add_errno = true);
// sets fd to nonblocking if <tt>set_nonblocking == true</tt> // sets fd to nonblocking if <tt>set_nonblocking == true</tt>
// or to blocking if <tt>set_nonblocking == false</tt> // or to blocking if <tt>set_nonblocking == false</tt>
// throws @p ios_base::failure on error // throws @p network_error on error
void nonblocking(native_socket_type fd, bool new_value); void nonblocking(native_socket_type fd, bool new_value);
// returns true if fd is nodelay socket // returns true if fd is nodelay socket
// throws @p ios_base::failure on error // throws @p network_error on error
void tcp_nodelay(native_socket_type fd, bool new_value); void tcp_nodelay(native_socket_type fd, bool new_value);
// reads @p result and @p errno and throws @p ios_base::failure on error // reads @p result and @p errno and throws @p network_error on error
void handle_write_result(ssize_t result, bool is_nonblocking_io); void handle_write_result(ssize_t result, bool is_nonblocking_io);
// reads @p result and @p errno and throws @p ios_base::failure on error // reads @p result and @p errno and throws @p network_error on error
void handle_read_result(ssize_t result, bool is_nonblocking_io); void handle_read_result(ssize_t result, bool is_nonblocking_io);
std::pair<native_socket_type, native_socket_type> create_pipe(); std::pair<native_socket_type, native_socket_type> create_pipe();
......
...@@ -104,8 +104,6 @@ class network_error : public cppa_exception { ...@@ -104,8 +104,6 @@ class network_error : public cppa_exception {
~network_error(); ~network_error();
network_error(std::string&& what_str); network_error(std::string&& what_str);
network_error(const std::string& what_str); network_error(const std::string& what_str);
network_error(const network_error&) = default;
network_error& operator=(const network_error&) = default;
}; };
...@@ -135,6 +133,23 @@ class bind_failure : public network_error { ...@@ -135,6 +133,23 @@ class bind_failure : public network_error {
}; };
/**
* @brief Thrown to indicate that a client tried to read from
* a stream after is has been shutdown by the remote side.
*/
class stream_at_eof : public network_error {
using super = network_error;
public:
~stream_at_eof();
template<typename... Ts>
stream_at_eof(Ts&&... args) : super(std::forward<Ts>(args)...) { }
};
inline std::uint32_t actor_exited::reason() const noexcept { inline std::uint32_t actor_exited::reason() const noexcept {
return m_reason; return m_reason;
} }
......
...@@ -45,14 +45,14 @@ class input_stream : public virtual ref_counted { ...@@ -45,14 +45,14 @@ class input_stream : public virtual ref_counted {
/** /**
* @brief Reads exactly @p num_bytes from the data source and blocks the * @brief Reads exactly @p num_bytes from the data source and blocks the
* caller if needed. * caller if needed.
* @throws std::ios_base::failure * @throws network_error
*/ */
virtual void read(void* buf, size_t num_bytes) = 0; virtual void read(void* buf, size_t num_bytes) = 0;
/** /**
* @brief Tries to read up to @p num_bytes from the data source. * @brief Tries to read up to @p num_bytes from the data source.
* @returns The number of read bytes. * @returns The number of read bytes.
* @throws std::ios_base::failure * @throws network_error
*/ */
virtual size_t read_some(void* buf, size_t num_bytes) = 0; virtual size_t read_some(void* buf, size_t num_bytes) = 0;
......
...@@ -46,14 +46,14 @@ class output_stream : public virtual ref_counted { ...@@ -46,14 +46,14 @@ class output_stream : public virtual ref_counted {
* @brief Writes @p num_bytes bytes from @p buf to the data sink. * @brief Writes @p num_bytes bytes from @p buf to the data sink.
* @note This member function blocks until @p num_bytes were successfully * @note This member function blocks until @p num_bytes were successfully
* written. * written.
* @throws std::ios_base::failure * @throws network_error
*/ */
virtual void write(const void* buf, size_t num_bytes) = 0; virtual void write(const void* buf, size_t num_bytes) = 0;
/** /**
* @brief Tries to write up to @p num_bytes bytes from @p buf. * @brief Tries to write up to @p num_bytes bytes from @p buf.
* @returns The number of written bytes. * @returns The number of written bytes.
* @throws std::ios_base::failure * @throws network_error
*/ */
virtual size_t write_some(const void* buf, size_t num_bytes) = 0; virtual size_t write_some(const void* buf, size_t num_bytes) = 0;
......
...@@ -184,7 +184,11 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -184,7 +184,11 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
} }
auto before = buf.size(); auto before = buf.size();
try { buf.append_from(m_in.get()); } try { buf.append_from(m_in.get()); }
catch (std::ios_base::failure&) { catch (stream_at_eof&) {
disconnect();
return continue_reading_result::closed;
}
catch (network_error&) {
disconnect(); disconnect();
return continue_reading_result::failure; return continue_reading_result::failure;
} }
...@@ -246,23 +250,21 @@ class broker::doorman : public broker::servant { ...@@ -246,23 +250,21 @@ class broker::doorman : public broker::servant {
continue_reading_result continue_reading() override { continue_reading_result continue_reading() override {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
for (;;) { optional<stream_ptr_pair> opt{none};
optional<stream_ptr_pair> opt{none}; try { opt = m_ptr->try_accept_connection(); }
try { opt = m_ptr->try_accept_connection(); } catch (std::exception& e) {
catch (std::exception& e) { CPPA_LOG_ERROR(to_verbose_string(e));
CPPA_LOG_ERROR(to_verbose_string(e)); static_cast<void>(e); // keep compiler happy
static_cast<void>(e); // keep compiler happy return continue_reading_result::failure;
return continue_reading_result::failure; }
} if (opt) {
if (opt) { using namespace std;
using namespace std; auto& p = *opt;
auto& p = *opt; get_ref<0>(m_accept_msg).handle = m_broker->add_connection(move(p.first),
get_ref<0>(m_accept_msg).handle = m_broker->add_connection(move(p.first), move(p.second));
move(p.second)); m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg);
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg); }
} return continue_reading_result::continue_later;
else return continue_reading_result::continue_later;
}
} }
protected: protected:
...@@ -338,15 +340,14 @@ void broker::invoke_message(msg_hdr_cref hdr, any_tuple msg) { ...@@ -338,15 +340,14 @@ void broker::invoke_message(msg_hdr_cref hdr, any_tuple msg) {
m_dummy_node.sender = actor_addr{}; m_dummy_node.sender = actor_addr{};
m_dummy_node.msg.reset(); m_dummy_node.msg.reset();
// cleanup if needed // cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) { auto per = planned_exit_reason();
cleanup(planned_exit_reason()); if (bhvr_stack().empty() && per != exit_reason::not_exited) {
// release implicit reference count held by MM
deref();
}
else if (bhvr_stack().empty()) {
CPPA_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason"); CPPA_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason");
quit(exit_reason::normal); per = exit_reason::normal;
cleanup(planned_exit_reason()); quit(per);
}
if (per != exit_reason::not_exited) {
cleanup(per);
// release implicit reference count held by MM // release implicit reference count held by MM
deref(); deref();
} }
...@@ -386,6 +387,14 @@ broker::broker() : m_initialized(false) { ...@@ -386,6 +387,14 @@ broker::broker() : m_initialized(false) {
} }
void broker::cleanup(std::uint32_t reason) { void broker::cleanup(std::uint32_t reason) {
CPPA_LOG_TRACE(CPPA_ARG(reason));
auto mm = get_middleman();
for (auto& dm : m_accept) {
mm->stop_reader(dm.second.get());
}
for (auto& ds : m_io) {
mm->stop_reader(ds.second.get());
}
super::cleanup(reason); super::cleanup(reason);
get_actor_registry()->dec_running(); get_actor_registry()->dec_running();
} }
......
...@@ -18,9 +18,9 @@ ...@@ -18,9 +18,9 @@
#include <atomic> #include <atomic>
#include <ios> // std::ios_base::failure
#include <cstring> #include <cstring>
#include <utility> #include <utility>
#include <stdexcept> // invalid_argument
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
#include "cppa/util/buffer.hpp" #include "cppa/util/buffer.hpp"
...@@ -135,7 +135,7 @@ void buffer::write(size_t num_bytes, const void* data, buffer_write_policy wp) { ...@@ -135,7 +135,7 @@ void buffer::write(size_t num_bytes, const void* data, buffer_write_policy wp) {
acquire(num_bytes); acquire(num_bytes);
} }
else if (num_bytes > remaining()) { else if (num_bytes > remaining()) {
throw std::ios_base::failure("final buffer size exceeded"); throw network_error("final buffer size exceeded");
} }
memcpy(wr_ptr(), data, num_bytes); memcpy(wr_ptr(), data, num_bytes);
inc_size(num_bytes); inc_size(num_bytes);
......
...@@ -99,4 +99,6 @@ bind_failure::bind_failure(int err_code) : network_error(be_what(err_code)) { ...@@ -99,4 +99,6 @@ bind_failure::bind_failure(int err_code) : network_error(be_what(err_code)) {
bind_failure::~bind_failure() { } bind_failure::~bind_failure() { }
stream_at_eof::~stream_at_eof() { }
} // namespace cppa } // namespace cppa
...@@ -17,7 +17,6 @@ ...@@ -17,7 +17,6 @@
\******************************************************************************/ \******************************************************************************/
#include <ios>
#include <sstream> #include <sstream>
#include <cstring> #include <cstring>
...@@ -54,9 +53,9 @@ void throw_io_failure [[noreturn]] (const char* what, bool add_errno) { ...@@ -54,9 +53,9 @@ void throw_io_failure [[noreturn]] (const char* what, bool add_errno) {
std::ostringstream oss; std::ostringstream oss;
oss << what << ": " << last_socket_error_as_string() oss << what << ": " << last_socket_error_as_string()
<< " [errno: " << last_socket_error() << "]"; << " [errno: " << last_socket_error() << "]";
throw std::ios_base::failure(oss.str()); throw network_error(oss.str());
} }
throw std::ios_base::failure(what); throw network_error(what);
} }
void tcp_nodelay(native_socket_type fd, bool new_value) { void tcp_nodelay(native_socket_type fd, bool new_value) {
...@@ -83,8 +82,8 @@ void handle_write_result(ssize_t res, bool is_nonblock) { ...@@ -83,8 +82,8 @@ void handle_write_result(ssize_t res, bool is_nonblock) {
} }
void handle_read_result(ssize_t res, bool is_nonblock) { void handle_read_result(ssize_t res, bool is_nonblock) {
handle_io_result(res, is_nonblock, "cannot read from file descriptor"); handle_io_result(res, is_nonblock, "cannot read from socket");
if (res == 0) throw_io_failure("cannot read from closed file descriptor"); if (res == 0) throw stream_at_eof("cannot read from closed socket");
} }
} // namespace fd_util } // namespace fd_util
......
...@@ -431,12 +431,12 @@ void middleman_loop(middleman_impl* impl) { ...@@ -431,12 +431,12 @@ void middleman_loop(middleman_impl* impl) {
case continue_writing_result::failure: case continue_writing_result::failure:
io->io_failed(event::write); io->io_failed(event::write);
impl->stop_writer(io); impl->stop_writer(io);
CPPA_LOGF_DEBUG("writer removed because " CPPA_LOGF_DEBUG("stopped writer because "
"of an error"); "of an error");
break; break;
case continue_writing_result::closed: case continue_writing_result::closed:
impl->stop_writer(io); impl->stop_writer(io);
CPPA_LOGF_DEBUG("writer removed because " CPPA_LOGF_DEBUG("stopped writer because "
"connection has been closed"); "connection has been closed");
break; break;
case continue_writing_result::done: case continue_writing_result::done:
...@@ -456,12 +456,12 @@ void middleman_loop(middleman_impl* impl) { ...@@ -456,12 +456,12 @@ void middleman_loop(middleman_impl* impl) {
case continue_reading_result::failure: case continue_reading_result::failure:
io->io_failed(event::read); io->io_failed(event::read);
impl->stop_reader(io); impl->stop_reader(io);
CPPA_LOGF_DEBUG("peer removed because a " CPPA_LOGF_DEBUG("stopped reader because "
"read error has occured"); "of an error");
break; break;
case continue_reading_result::closed: case continue_reading_result::closed:
impl->stop_reader(io); impl->stop_reader(io);
CPPA_LOGF_DEBUG("peer removed because " CPPA_LOGF_DEBUG("stopped reader because "
"connection has been closed"); "connection has been closed");
break; break;
case continue_reading_result::continue_later: case continue_reading_result::continue_later:
......
...@@ -130,17 +130,19 @@ void middleman_event_handler::update() { ...@@ -130,17 +130,19 @@ void middleman_event_handler::update() {
auto last = m_meta.end(); auto last = m_meta.end();
// checks whether an element can be safely deleted, // checks whether an element can be safely deleted,
// i.e., was not put back into m_meta by some alteration // i.e., was not put back into m_meta by some alteration
auto is_alive = [&](native_socket_type fd) -> bool { auto not_in_meta = [&](native_socket_type fd) -> bool {
auto iter = std::lower_bound(first, last, fd, mless); auto iter = std::lower_bound(first, last, fd, mless);
return iter != last && iter->fd == fd; return iter == last || iter->fd != fd;
}; };
// dispose everything that wasn't put back into m_meta again // dispose everything that wasn't put back into m_meta again
for (auto elem : m_dispose_list) { for (auto elem : m_dispose_list) {
CPPA_LOG_DEBUG("dispose " << elem);
auto rd = elem->read_handle(); auto rd = elem->read_handle();
auto wr = elem->write_handle(); auto wr = elem->write_handle();
if ( (rd == wr && !is_alive(rd)) if ( (rd == wr && not_in_meta(rd))
|| (rd != wr && !is_alive(rd) && !is_alive(wr))) { || (rd != wr && not_in_meta(rd) && not_in_meta(wr))) {
elem->dispose(); // dispose continuable
elem->dispose();
} }
} }
m_dispose_list.clear(); m_dispose_list.clear();
......
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#if defined(CPPA_LINUX) && !defined(CPPA_POLL_IMPL) #if defined(CPPA_LINUX) && !defined(CPPA_POLL_IMPL)
#include <ios>
#include <string> #include <string>
#include <vector> #include <vector>
...@@ -51,7 +50,7 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -51,7 +50,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
void init() { void init() {
m_epollfd = epoll_create1(EPOLL_CLOEXEC); m_epollfd = epoll_create1(EPOLL_CLOEXEC);
if (m_epollfd == -1) { if (m_epollfd == -1) {
throw std::ios_base::failure( std::string("epoll_create1: ") throw network_error( std::string("epoll_create1: ")
+ strerror(errno)); + strerror(errno));
} }
// handle at most 64 events at a time // handle at most 64 events at a time
......
...@@ -16,7 +16,6 @@ ...@@ -16,7 +16,6 @@
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt * * accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/ \******************************************************************************/
#include <ios>
#include <cstring> #include <cstring>
#include <errno.h> #include <errno.h>
#include <iostream> #include <iostream>
...@@ -79,6 +78,7 @@ bool accept_impl(stream_ptr_pair& result, ...@@ -79,6 +78,7 @@ bool accept_impl(stream_ptr_pair& result,
auto sfd = ::accept(fd, &addr, &addrlen); auto sfd = ::accept(fd, &addr, &addrlen);
if (sfd == invalid_socket) { if (sfd == invalid_socket) {
auto err = last_socket_error(); auto err = last_socket_error();
CPPA_LOGF_DEBUG("accept failed for reason " << err);
if (nonblocking && would_block_or_temporarily_unavailable(err)) { if (nonblocking && would_block_or_temporarily_unavailable(err)) {
// ok, try again // ok, try again
return false; return false;
...@@ -128,7 +128,7 @@ std::unique_ptr<acceptor> tcp_acceptor::create(std::uint16_t port, ...@@ -128,7 +128,7 @@ std::unique_ptr<acceptor> tcp_acceptor::create(std::uint16_t port,
if (bind(sockfd, (sockaddr*) &serv_addr, sizeof(serv_addr)) < 0) { if (bind(sockfd, (sockaddr*) &serv_addr, sizeof(serv_addr)) < 0) {
throw bind_failure(errno); throw bind_failure(errno);
} }
if (listen(sockfd, 10) != 0) { if (listen(sockfd, SOMAXCONN) != 0) {
throw network_error("listen() failed"); throw network_error("listen() failed");
} }
// ok, no exceptions so far // ok, no exceptions so far
......
...@@ -17,7 +17,6 @@ ...@@ -17,7 +17,6 @@
\******************************************************************************/ \******************************************************************************/
#include <ios>
#include <cstring> #include <cstring>
#include <errno.h> #include <errno.h>
#include <iostream> #include <iostream>
...@@ -45,9 +44,12 @@ namespace io { ...@@ -45,9 +44,12 @@ namespace io {
using namespace ::cppa::detail::fd_util; using namespace ::cppa::detail::fd_util;
tcp_io_stream::tcp_io_stream(native_socket_type fd) : m_fd(fd) { } tcp_io_stream::tcp_io_stream(native_socket_type fd) : m_fd(fd) {
CPPA_LOG_TRACE("new IO stream using socket " << fd);
}
tcp_io_stream::~tcp_io_stream() { tcp_io_stream::~tcp_io_stream() {
CPPA_LOG_TRACE("close socket " << m_fd);
closesocket(m_fd); closesocket(m_fd);
} }
......
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include <ios> // ios_base::failure
#include <list> #include <list>
#include <memory> #include <memory>
#include <cstring> // memset #include <cstring> // memset
......
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