Unverified Commit 696736e4 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #731

Call remove_from_loop in manager::detach
parents 2cae44fc e932cc00
......@@ -72,8 +72,6 @@ public:
virtual void remove_endpoint(datagram_handle hdl) = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
bool consume(execution_unit*, datagram_handle hdl,
network::receive_buffer& buf) override;
......
......@@ -43,8 +43,6 @@ public:
~doorman() override;
void io_failure(execution_unit* ctx, network::operation op) override;
using doorman_base::new_connection;
bool new_connection(execution_unit* ctx, connection_handle x);
......
......@@ -167,6 +167,12 @@ public:
/// Get the next id to create a new datagram handle
int64_t next_endpoint_id();
/// Returns the number of socket handlers.
size_t num_socket_handlers() const noexcept;
/// Run all pending events generated from calls to `add` or `del`.
void handle_internal_events();
private:
/// Calls `epoll`, `kqueue`, or `poll` with or without blocking.
bool poll_once_impl(bool block);
......
......@@ -63,8 +63,8 @@ public:
/// Adds the I/O device to the event loop of the middleman.
virtual void add_to_loop() = 0;
/// Called by the underlying I/O device to report failures.
virtual void io_failure(execution_unit* ctx, operation op) = 0;
/// Detaches this manager from its parent in case of an error.
void io_failure(execution_unit* ctx, operation op);
/// Get the address of the underlying I/O device.
virtual std::string addr() const = 0;
......
......@@ -57,8 +57,6 @@ public:
/// content of the buffer via the network.
virtual void flush() = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
bool consume(execution_unit*, const void*, size_t) override;
void data_transferred(execution_unit*, size_t, size_t) override;
......
......@@ -73,13 +73,6 @@ void datagram_servant::datagram_sent(execution_unit* ctx, datagram_handle hdl,
invoke_mailbox_element_impl(ctx, tmp);
}
void datagram_servant::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(ctx, true);
}
} // namespace io
} // namespace caf
......@@ -205,10 +205,7 @@ namespace network {
auto fd = ptr ? ptr->fd() : pipe_.first;
handle_socket_event(fd, static_cast<int>(iter->events), ptr);
}
for (auto& me : events_) {
handle(me);
}
events_.clear();
handle_internal_events();
return true;
}
}
......@@ -285,6 +282,10 @@ namespace network {
}
}
size_t default_multiplexer::num_socket_handlers() const noexcept {
return shadow_;
}
#else // CAF_EPOLL_MULTIPLEXER
// Let's be honest: the API of poll() sucks. When dealing with 1000 sockets
......@@ -383,11 +384,8 @@ namespace network {
// operations possible on the socket
handle_socket_event(e.fd, e.mask, e.ptr);
}
CAF_LOG_DEBUG(CAF_ARG(events_.size()));
poll_res.clear();
for (auto& me : events_)
handle(me);
events_.clear();
handle_internal_events();
return true;
}
}
......@@ -456,6 +454,10 @@ namespace network {
}
}
size_t default_multiplexer::num_socket_handlers() const noexcept {
return pollset_.size();
}
#endif // CAF_EPOLL_MULTIPLEXER
// -- Helper functions for defining bitmasks of event handlers -----------------
......@@ -604,9 +606,7 @@ bool default_multiplexer::poll_once(bool block) {
internally_posted_.swap(xs);
for (auto& ptr : xs)
resume(std::move(ptr));
for (auto& me : events_)
handle(me);
events_.clear();
handle_internal_events();
// Try to swap back to internall_posted_ to re-use allocated memory.
if (internally_posted_.empty()) {
xs.swap(internally_posted_);
......@@ -736,6 +736,13 @@ int64_t default_multiplexer::next_endpoint_id() {
return servant_ids_++;
}
void default_multiplexer::handle_internal_events() {
CAF_LOG_TRACE(CAF_ARG2("num-events", events_.size()));
for (auto& e : events_)
handle(e);
events_.clear();
}
// -- Related helper functions -------------------------------------------------
template <int Family>
......
......@@ -37,13 +37,6 @@ message doorman::detach_message() {
return make_message(acceptor_closed_msg{hdl()});
}
void doorman::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl().id()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(ctx, true);
}
bool doorman::new_connection(execution_unit* ctx, connection_handle x) {
msg().handle = x;
return invoke_mailbox_element(ctx);
......
......@@ -43,11 +43,16 @@ abstract_broker* manager::parent() {
}
void manager::detach(execution_unit*, bool invoke_disconnect_message) {
CAF_LOG_TRACE("");
CAF_LOG_TRACE(CAF_ARG(invoke_disconnect_message));
// This function gets called from the multiplexer when an error occurs or
// from the broker when closing this manager. In both cases, we need to make
// sure this manager does not receive further socket events.
remove_from_loop();
// Disconnect from the broker if not already detached.
if (!detached()) {
CAF_LOG_DEBUG("disconnect servant from broker");
auto raw_ptr = parent();
// keep the strong reference until we go out of scope
// Keep a strong reference to our parent until we go out of scope.
strong_actor_ptr ptr;
ptr.swap(parent_);
detach_from(raw_ptr);
......@@ -69,6 +74,12 @@ void manager::detach(execution_unit*, bool invoke_disconnect_message) {
}
}
void manager::io_failure(execution_unit* ctx, operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
CAF_IGNORE_UNUSED(op);
detach(ctx, true);
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -77,12 +77,5 @@ void scribe::data_transferred(execution_unit* ctx, size_t written,
//parent()->consume(std::move(ptr));
}
void scribe::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(ctx, true);
}
} // namespace io
} // namespace caf
......@@ -196,8 +196,6 @@ void stream::handle_error_propagation() {
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
} // namespace network
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE io_default_multiplexer
#include "caf/test/io_dsl.hpp"
#include <vector>
#include <algorithm>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/operation.hpp"
using namespace caf;
namespace {
struct sub_fixture : test_coordinator_fixture<> {
io::network::default_multiplexer mpx;
sub_fixture() : mpx(&sys) {
// nop
}
bool exec_all() {
size_t count = 0;
while (mpx.poll_once(false)) {
++count;
}
return count != 0;
}
};
struct fixture {
sub_fixture client;
sub_fixture server;
void exec_all() {
while (client.exec_all() || server.exec_all()) {
// Rince and repeat.
}
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(default_multiplexer_tests, fixture)
CAF_TEST(doorman io_failure) {
CAF_MESSAGE("add doorman to server");
// The multiplexer adds a pipe reader on startup.
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 1u);
auto doorman = unbox(server.mpx.new_tcp_doorman(0, nullptr, false));
doorman->add_to_loop();
server.mpx.handle_internal_events();
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 2u);
CAF_MESSAGE("trigger I/O failure in doorman");
doorman->io_failure(&server.mpx, io::network::operation::propagate_error);
server.mpx.handle_internal_events();
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 1u);
}
CAF_TEST(scribe io_failure) {
CAF_MESSAGE("add doorman to server");
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 1u);
auto doorman = unbox(server.mpx.new_tcp_doorman(0, nullptr, false));
doorman->add_to_loop();
server.mpx.handle_internal_events();
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 2u);
CAF_MESSAGE("connect to server (add scribe to client)");
auto scribe = unbox(client.mpx.new_tcp_scribe("localhost", doorman->port()));
CAF_CHECK_EQUAL(client.mpx.num_socket_handlers(), 1u);
scribe->add_to_loop();
client.mpx.handle_internal_events();
CAF_CHECK_EQUAL(client.mpx.num_socket_handlers(), 2u);
CAF_MESSAGE("trigger I/O failure in scribe");
scribe->io_failure(&client.mpx, io::network::operation::propagate_error);
client.mpx.handle_internal_events();
CAF_CHECK_EQUAL(client.mpx.num_socket_handlers(), 1u);
CAF_MESSAGE("trigger I/O failure in doorman");
doorman->io_failure(&server.mpx, io::network::operation::propagate_error);
server.mpx.handle_internal_events();
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 1u);
}
CAF_TEST_FIXTURE_SCOPE_END()
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