Commit 4d140253 authored by Joseph Noir's avatar Joseph Noir

Adopt graceful_shutdown for event handlers

parent f9864512
...@@ -152,6 +152,15 @@ struct newb : public network::newb_base { ...@@ -152,6 +152,15 @@ struct newb : public network::newb_base {
intrusive_ptr_release(this->ctrl()); intrusive_ptr_release(this->ctrl());
} }
void graceful_shutdown() override {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Ignore repeated calls.
if (state_.shutting_down)
return;
state_.shutting_down = true;
trans->shutdown(this, fd_);
}
// -- base requirements ------------------------------------------------------ // -- base requirements ------------------------------------------------------
void start() override { void start() override {
...@@ -169,9 +178,7 @@ struct newb : public network::newb_base { ...@@ -169,9 +178,7 @@ struct newb : public network::newb_base {
void stop() override { void stop() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
close_read_channel(); graceful_shutdown();
stop_reading();
stop_writing();
} }
void io_error(network::operation op, error err) override { void io_error(network::operation op, error err) override {
...@@ -193,7 +200,7 @@ struct newb : public network::newb_base { ...@@ -193,7 +200,7 @@ struct newb : public network::newb_base {
} }
switch (op) { switch (op) {
case operation::read: case operation::read:
stop_reading(); graceful_shutdown();
break; break;
case operation::write: case operation::write:
stop_writing(); stop_writing();
...@@ -217,13 +224,13 @@ struct newb : public network::newb_base { ...@@ -217,13 +224,13 @@ struct newb : public network::newb_base {
void start_writing() override { void start_writing() override {
if (!writing_) { if (!writing_) {
event_handler::backend().add(network::operation::write, fd(), this); event_handler::backend().add(network::operation::write, fd_, this);
writing_ = true; writing_ = true;
} }
} }
void stop_writing() override { void stop_writing() override {
event_handler::backend().del(network::operation::write, fd(), this); event_handler::backend().del(network::operation::write, fd_, this);
} }
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
...@@ -468,6 +475,15 @@ struct newb_acceptor : network::newb_base { ...@@ -468,6 +475,15 @@ struct newb_acceptor : network::newb_base {
intrusive_ptr_release(this->ctrl()); intrusive_ptr_release(this->ctrl());
} }
void graceful_shutdown() override {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Ignore repeated calls.
if (state_.shutting_down)
return;
state_.shutting_down = true;
accept_pol->shutdown(this, fd_);
}
// -- base requirements ------------------------------------------------------ // -- base requirements ------------------------------------------------------
void start() override { void start() override {
...@@ -485,9 +501,7 @@ struct newb_acceptor : network::newb_base { ...@@ -485,9 +501,7 @@ struct newb_acceptor : network::newb_base {
void stop() override { void stop() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
close_read_channel(); graceful_shutdown();
stop_reading();
stop_writing();
} }
void io_error(network::operation op, error err) override { void io_error(network::operation op, error err) override {
...@@ -511,13 +525,13 @@ struct newb_acceptor : network::newb_base { ...@@ -511,13 +525,13 @@ struct newb_acceptor : network::newb_base {
void start_writing() override { void start_writing() override {
if (!writing_) { if (!writing_) {
event_handler::backend().add(network::operation::write, fd(), this); event_handler::backend().add(network::operation::write, fd_, this);
writing_ = true; writing_ = true;
} }
} }
void stop_writing() override { void stop_writing() override {
event_handler::backend().del(network::operation::write, fd(), this); event_handler::backend().del(network::operation::write, fd_, this);
} }
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
......
...@@ -19,8 +19,9 @@ ...@@ -19,8 +19,9 @@
#pragma once #pragma once
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/policy/transport.hpp" #include "caf/expected.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/policy/transport.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -68,6 +69,11 @@ struct accept { ...@@ -68,6 +69,11 @@ struct accept {
// nop // nop
} }
virtual void shutdown(io::network::newb_base*,
io::network::native_socket) {
// nop
}
bool manual_read; bool manual_read;
}; };
......
...@@ -48,6 +48,8 @@ struct tcp_transport : public transport { ...@@ -48,6 +48,8 @@ struct tcp_transport : public transport {
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override; optional<io::network::protocol::network> preferred = none) override;
void shutdown(io::network::newb_base*, io::network::native_socket) override;
// State for reading. // State for reading.
size_t read_threshold; size_t read_threshold;
size_t collected; size_t collected;
...@@ -81,6 +83,11 @@ struct accept_tcp : public accept<Message> { ...@@ -81,6 +83,11 @@ struct accept_tcp : public accept<Message> {
void init(io::network::newb_base*, io::newb<Message>& spawned) override { void init(io::network::newb_base*, io::newb<Message>& spawned) override {
spawned.start(); spawned.start();
} }
void shutdown(io::network::newb_base*,
io::network::native_socket sockfd) override {
io::network::shutdown_both(sockfd);
}
}; };
template <class T> template <class T>
......
...@@ -55,6 +55,9 @@ struct udp_transport : public transport { ...@@ -55,6 +55,9 @@ struct udp_transport : public transport {
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override; optional<io::network::protocol::network> preferred = none) override;
void shutdown(io::network::newb_base* parent,
io::network::native_socket sockfd) override;
// State for reading. // State for reading.
size_t maximum; size_t maximum;
bool first_message; bool first_message;
...@@ -100,6 +103,11 @@ struct accept_udp : public accept<Message> { ...@@ -100,6 +103,11 @@ struct accept_udp : public accept<Message> {
spawned.trans->read_some(parent, *spawned.proto.get()); spawned.trans->read_some(parent, *spawned.proto.get());
spawned.start(); spawned.start();
} }
void shutdown(io::network::newb_base* parent,
io::network::native_socket) override {
parent->passivate();
}
}; };
template <class T> template <class T>
......
/****************************************************************************** /******************************************************************************
* ____ _ _____ * * ____ _ _____ *
* / ___| / \ | ___| C++ * * / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor * * | | / _ \ | |_ Actor * | |___ / ___ \| _| Framework *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2018 Dominik Charousset *
...@@ -96,6 +95,9 @@ struct transport { ...@@ -96,6 +95,9 @@ struct transport {
connect(const std::string&, uint16_t, connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none); optional<io::network::protocol::network> = none);
virtual void shutdown(io::network::newb_base* parent,
io::network::native_socket sockfd);
size_t received_bytes; size_t received_bytes;
size_t max_consecutive_reads; size_t max_consecutive_reads;
......
...@@ -179,6 +179,11 @@ tcp_transport::connect(const std::string& host, uint16_t port, ...@@ -179,6 +179,11 @@ tcp_transport::connect(const std::string& host, uint16_t port,
return io::network::new_tcp_connection(host, port, preferred); return io::network::new_tcp_connection(host, port, preferred);
} }
void tcp_transport::shutdown(io::network::newb_base*,
io::network::native_socket sockfd) {
io::network::shutdown_both(sockfd);
}
io::network::native_socket get_newb_socket(io::network::newb_base* n) { io::network::native_socket get_newb_socket(io::network::newb_base* n) {
return n->fd(); return n->fd();
} }
......
...@@ -181,5 +181,10 @@ udp_transport::connect(const std::string& host, uint16_t port, ...@@ -181,5 +181,10 @@ udp_transport::connect(const std::string& host, uint16_t port,
return res->first; return res->first;
} }
void udp_transport::shutdown(io::network::newb_base* parent,
io::network::native_socket) {
parent->stop_reading();
}
} // namespace policy } // namespace policy
} // namespace caf } // namespace caf
...@@ -73,10 +73,15 @@ byte_buffer& transport::wr_buf() { ...@@ -73,10 +73,15 @@ byte_buffer& transport::wr_buf() {
expected<io::network::native_socket> expected<io::network::native_socket>
transport::connect(const std::string&, uint16_t, transport::connect(const std::string&, uint16_t,
optional<io::network::protocol::network>) { optional<io::network::protocol::network>) {
return sec::bad_function_call; return sec::bad_function_call;
} }
void transport::shutdown(io::network::newb_base*,
io::network::native_socket sockfd) {
io::network::shutdown_both(sockfd);
}
// -- protocol_policy_base ----------------------------------------------------- // -- protocol_policy_base -----------------------------------------------------
protocol_base::~protocol_base() { protocol_base::~protocol_base() {
......
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