Commit fb38580f authored by Joseph Noir's avatar Joseph Noir

Fix shutdown to await multiplexer events

parent b5ab8111
...@@ -83,6 +83,15 @@ struct newb_base : public network::event_handler { ...@@ -83,6 +83,15 @@ struct newb_base : public network::event_handler {
virtual void stop() = 0; virtual void stop() = 0;
virtual void io_error(operation op, error err) = 0; virtual void io_error(operation op, error err) = 0;
virtual void start_reading() = 0;
virtual void stop_reading() = 0;
virtual void start_writing() = 0;
virtual void stop_writing() = 0;
}; };
// -- transport policy --------------------------------------------------------- // -- transport policy ---------------------------------------------------------
...@@ -296,6 +305,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -296,6 +305,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
return resumable::io_actor; return resumable::io_actor;
} }
const char* name() const override {
return "newb";
}
// -- overridden modifiers of local_actor ------------------------------------ // -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override { void launch(execution_unit* eu, bool lazy, bool hide) override {
...@@ -327,8 +340,8 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -327,8 +340,8 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
bool cleanup(error&& reason, execution_unit* host) override { bool cleanup(error&& reason, execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
stop(); // ???
// TODO: Should policies be notified here? // TODO: Should policies be notified here?
stop();
return local_actor::cleanup(std::move(reason), host); return local_actor::cleanup(std::move(reason), host);
} }
...@@ -364,26 +377,27 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -364,26 +377,27 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
switch (op) { switch (op) {
case network::operation::read: break; case network::operation::read:
case network::operation::write: break; reading_ = false;
break;
case network::operation::write:
writing_ = false;
break;
case network::operation::propagate_error: ; // nop case network::operation::propagate_error: ; // nop
} }
// nop // Quit if there is nothing left to do.
if (!reading_ && !writing_)
intrusive_ptr_release(super::ctrl());
} }
// -- members ---------------------------------------------------------------- // -- base requirements ------------------------------------------------------
void init_newb() {
CAF_LOG_TRACE("");
super::setf(super::is_initialized_flag);
}
void start() override { void start() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_add_ref(super::ctrl()); intrusive_ptr_add_ref(super::ctrl());
CAF_LOG_DEBUG("starting newb"); CAF_LOG_DEBUG("starting newb");
event_handler::activate(); start_reading();
if (transport) if (transport)
transport->prepare_next_read(this); transport->prepare_next_read(this);
} }
...@@ -391,12 +405,14 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -391,12 +405,14 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
void stop() override { void stop() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_release(super::ctrl());
close_read_channel(); close_read_channel();
passivate(); stop_reading();
stop_writing();
} }
void io_error(operation op, error err) override { void io_error(operation op, error err) override {
// TODO: This message is not always handled correctly.
// Don't know why yet ...
auto mptr = make_mailbox_element(nullptr, invalid_message_id, {}, auto mptr = make_mailbox_element(nullptr, invalid_message_id, {},
io_error_msg{op, std::move(err)}); io_error_msg{op, std::move(err)});
switch (scheduled_actor::consume(*mptr)) { switch (scheduled_actor::consume(*mptr)) {
...@@ -412,10 +428,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -412,10 +428,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
} }
switch (op) { switch (op) {
case operation::read: case operation::read:
passivate(); stop_reading();
break; break;
case operation::write: case operation::write:
event_handler::backend().del(operation::write, fd(), this); stop_writing();
break; break;
case operation::propagate_error: case operation::propagate_error:
// TODO: What should happen here? // TODO: What should happen here?
...@@ -423,6 +439,35 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -423,6 +439,35 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
} }
} }
void start_reading() override {
if (!reading_) {
activate();
reading_ = true;
}
}
void stop_reading() override {
passivate();
}
void start_writing() override {
if (!writing_) {
event_handler::backend().add(operation::write, fd(), this);
writing_ = true;
}
}
void stop_writing() override {
event_handler::backend().del(operation::write, fd(), this);
}
// -- members ----------------------------------------------------------------
void init_newb() {
CAF_LOG_TRACE("");
super::setf(super::is_initialized_flag);
}
/// Get a write buffer to write data to be sent by this broker. /// Get a write buffer to write data to be sent by this broker.
write_handle<Message> wr_buf(header_writer* hw) { write_handle<Message> wr_buf(header_writer* hw) {
auto& buf = transport->wr_buf(); auto& buf = transport->wr_buf();
...@@ -515,12 +560,14 @@ private: ...@@ -515,12 +560,14 @@ private:
} }
mailbox_element_vals<Message> value_; mailbox_element_vals<Message> value_;
bool reading_;
bool writing_;
}; };
// -- new broker acceptor ------------------------------------------------------ // -- new broker acceptor ------------------------------------------------------
template <class Message> template <class Message>
struct newb_acceptor : public newb_base { struct newb_acceptor : public newb_base, public caf::ref_counted {
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
...@@ -529,6 +576,10 @@ struct newb_acceptor : public newb_base { ...@@ -529,6 +576,10 @@ struct newb_acceptor : public newb_base {
// nop // nop
} }
~newb_acceptor() {
// nop
}
// -- overridden modifiers of event handler ---------------------------------- // -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override { void handle_event(network::operation op) override {
...@@ -549,11 +600,61 @@ struct newb_acceptor : public newb_base { ...@@ -549,11 +600,61 @@ struct newb_acceptor : public newb_base {
void removed_from_loop(network::operation op) override { void removed_from_loop(network::operation op) override {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
CAF_LOG_DEBUG("newb removed from loop: " << to_string(op));
switch (op) { switch (op) {
case network::operation::read: break; case network::operation::read:
case network::operation::write: break; reading_ = false;
break;
case network::operation::write:
writing_ = false;
break;
case network::operation::propagate_error: ; // nop case network::operation::propagate_error: ; // nop
} }
// Quit if there is nothing left to do.
if (!reading_ && !writing_)
deref();
}
// -- base requirements ------------------------------------------------------
void start() override {
start_reading();
ref();
}
void stop() override {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
stop_reading();
stop_writing();
}
void io_error(operation op, error err) override {
std::cerr << "operation " << to_string(op) << " failed: "
<< backend().system().render(err) << std::endl;
stop();
}
void start_reading() override {
if (!reading_) {
activate();
reading_ = true;
}
}
void stop_reading() override {
passivate();
}
void start_writing() override {
if (!writing_) {
event_handler::backend().add(operation::write, fd(), this);
writing_ = true;
}
}
void stop_writing() override {
event_handler::backend().del(operation::write, fd(), this);
} }
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
...@@ -577,27 +678,19 @@ struct newb_acceptor : public newb_base { ...@@ -577,27 +678,19 @@ struct newb_acceptor : public newb_base {
acceptor->multiplex_write(); acceptor->multiplex_write();
} }
void start() override {
event_handler::activate();
}
void stop() override {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
void io_error(operation op, error err) override {
std::cerr << "operation " << to_string(op) << " failed: "
<< backend().system().render(err) << std::endl;
}
virtual expected<actor> create_newb(native_socket sock, virtual expected<actor> create_newb(native_socket sock,
transport_policy_ptr pol) = 0; transport_policy_ptr pol) = 0;
std::unique_ptr<accept_policy> acceptor; std::unique_ptr<accept_policy> acceptor;
private:
bool reading_;
bool writing_;
}; };
template <class T>
using acceptor_ptr = caf::intrusive_ptr<newb_acceptor<T>>;
// -- factories ---------------------------------------------------------------- // -- factories ----------------------------------------------------------------
// Goal: // Goal:
...@@ -716,7 +809,7 @@ actor make_client_newb(actor_system& sys, std::string host, uint16_t port) { ...@@ -716,7 +809,7 @@ actor make_client_newb(actor_system& sys, std::string host, uint16_t port) {
} }
template <class NewbAcceptor, class AcceptPolicy> template <class NewbAcceptor, class AcceptPolicy>
std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys, caf::intrusive_ptr<NewbAcceptor> make_server_newb(actor_system& sys,
uint16_t port, uint16_t port,
const char* addr = nullptr, const char* addr = nullptr,
bool reuse_addr = false) { bool reuse_addr = false) {
...@@ -728,7 +821,7 @@ std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys, ...@@ -728,7 +821,7 @@ std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys,
return nullptr; return nullptr;
} }
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend()); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<NewbAcceptor> ptr{new NewbAcceptor(mpx, *esock)}; auto ptr = caf::make_counted<NewbAcceptor>(mpx, *esock);
ptr->acceptor = std::move(acc); ptr->acceptor = std::move(acc);
ptr->start(); ptr->start();
return ptr; return ptr;
......
...@@ -154,7 +154,7 @@ void tcp_transport::prepare_next_write(io::network::newb_base* parent) { ...@@ -154,7 +154,7 @@ void tcp_transport::prepare_next_write(io::network::newb_base* parent) {
written = 0; written = 0;
send_buffer.clear(); send_buffer.clear();
if (offline_buffer.empty()) { if (offline_buffer.empty()) {
parent->backend().del(io::network::operation::write, parent->fd(), parent); parent->stop_writing();
writing = false; writing = false;
} else { } else {
send_buffer.swap(offline_buffer); send_buffer.swap(offline_buffer);
...@@ -165,7 +165,7 @@ void tcp_transport::flush(io::network::newb_base* parent) { ...@@ -165,7 +165,7 @@ void tcp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write, parent->fd(), parent); parent->start_writing();
writing = true; writing = true;
prepare_next_write(parent); prepare_next_write(parent);
} }
......
...@@ -134,8 +134,7 @@ void udp_transport::prepare_next_write(io::network::newb_base* parent) { ...@@ -134,8 +134,7 @@ void udp_transport::prepare_next_write(io::network::newb_base* parent) {
send_sizes.clear(); send_sizes.clear();
if (offline_buffer.empty()) { if (offline_buffer.empty()) {
writing = false; writing = false;
parent->backend().del(io::network::operation::write, parent->stop_writing();
parent->fd(), parent);
} else { } else {
// Add size of last chunk. // Add size of last chunk.
offline_sizes.push_back(offline_buffer.size() - offline_sum); offline_sizes.push_back(offline_buffer.size() - offline_sum);
...@@ -160,8 +159,7 @@ void udp_transport::flush(io::network::newb_base* parent) { ...@@ -160,8 +159,7 @@ void udp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write, parent->start_writing();
parent->fd(), parent);
writing = true; writing = true;
prepare_next_write(parent); prepare_next_write(parent);
} }
......
...@@ -206,7 +206,7 @@ struct fixture { ...@@ -206,7 +206,7 @@ struct fixture {
default_multiplexer& mpx; default_multiplexer& mpx;
scheduler::test_coordinator& sched; scheduler::test_coordinator& sched;
actor self; actor self;
std::unique_ptr<acceptor_t> na; caf::intrusive_ptr<acceptor_t> na;
fixture() fixture()
: sys(cfg.parse(test::engine::argc(), test::engine::argv())), : sys(cfg.parse(test::engine::argc(), test::engine::argv())),
...@@ -224,7 +224,7 @@ struct fixture { ...@@ -224,7 +224,7 @@ struct fixture {
CAF_REQUIRE(esock); CAF_REQUIRE(esock);
// Create acceptor. // Create acceptor.
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend()); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<acceptor_t> ptr{new acceptor_t(mpx, esock->first)}; auto ptr = make_counted<acceptor_t>(mpx, esock->first);
ptr->acceptor.reset(new accept_policy_impl); ptr->acceptor.reset(new accept_policy_impl);
na = std::move(ptr); na = std::move(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