Commit 89b17d73 authored by Dominik Charousset's avatar Dominik Charousset

Implement graceful shutdown of TCP connections

parent e5057c0a
...@@ -286,7 +286,7 @@ public: ...@@ -286,7 +286,7 @@ public:
auto x = by_id(hdl); auto x = by_id(hdl);
if (!x) if (!x)
return false; return false;
x->stop_reading(); x->graceful_shutdown();
return true; return true;
} }
......
...@@ -57,11 +57,10 @@ public: ...@@ -57,11 +57,10 @@ public:
/// Activates the acceptor. /// Activates the acceptor.
void activate(acceptor_manager* mgr); void activate(acceptor_manager* mgr);
/// Closes the network connection and removes this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void graceful_shutdown() override;
protected: protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
......
...@@ -92,6 +92,8 @@ public: ...@@ -92,6 +92,8 @@ public:
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void graceful_shutdown() override;
void add_endpoint(datagram_handle hdl, const ip_endpoint& ep, void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr); const manager_ptr mgr);
......
...@@ -48,7 +48,7 @@ public: ...@@ -48,7 +48,7 @@ public:
network::receive_buffer& rd_buf() override; network::receive_buffer& rd_buf() override;
void stop_reading() override; void graceful_shutdown() override;
void flush() override; void flush() override;
......
...@@ -37,7 +37,7 @@ public: ...@@ -37,7 +37,7 @@ public:
bool new_connection() override; bool new_connection() override;
void stop_reading() override; void graceful_shutdown() override;
void launch() override; void launch() override;
......
...@@ -30,10 +30,21 @@ namespace network { ...@@ -30,10 +30,21 @@ namespace network {
/// A socket I/O event handler. /// A socket I/O event handler.
class event_handler { class event_handler {
public: public:
/// Stores various status flags and user-defined config parameters.
struct state { struct state {
/// Stores whether the socket is currently registered for reading.
bool reading : 1; bool reading : 1;
/// Stores whether the socket is currently registered for writing.
bool writing : 1; bool writing : 1;
/// Stores whether the parent actor demanded write receipts.
bool ack_writes : 1; bool ack_writes : 1;
/// Stores whether graceful_shutdown() was called.
bool shutting_down : 1;
/// Stores what receive policy is currently active.
receive_policy_flag rd_flag : 2; receive_policy_flag rd_flag : 2;
}; };
...@@ -50,6 +61,10 @@ public: ...@@ -50,6 +61,10 @@ public:
/// from the event loop for operations of type `op`. /// from the event loop for operations of type `op`.
virtual void removed_from_loop(operation op) = 0; virtual void removed_from_loop(operation op) = 0;
/// Shuts down communication on the managed socket, eventually removing
/// this event handler from the I/O loop.
virtual void graceful_shutdown() = 0;
/// Returns the native socket handle for this handler. /// Returns the native socket handle for this handler.
native_socket fd() const { native_socket fd() const {
return fd_; return fd_;
...@@ -75,16 +90,9 @@ public: ...@@ -75,16 +90,9 @@ public:
return !state_.reading; return !state_.reading;
} }
/// Closes the read channel of the underlying socket.
void close_read_channel();
/// Removes the file descriptor from the event loop of the parent. /// Removes the file descriptor from the event loop of the parent.
void passivate(); void passivate();
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
/// Returns whether this event handlers signals successful writes to its /// Returns whether this event handlers signals successful writes to its
/// parent actor. /// parent actor.
bool ack_writes() { bool ack_writes() {
......
...@@ -53,9 +53,8 @@ public: ...@@ -53,9 +53,8 @@ public:
/// if `invoke_detach_message == true`. /// if `invoke_detach_message == true`.
void detach(execution_unit* ctx, bool invoke_disconnect_message); void detach(execution_unit* ctx, bool invoke_disconnect_message);
/// Causes the manager to stop read operations on its I/O device. /// Causes the manager to gracefully close its connection.
/// Unwritten bytes are still send before the socket will be closed. virtual void graceful_shutdown() = 0;
virtual void stop_reading() = 0;
/// Removes the I/O device to the event loop of the middleman. /// Removes the I/O device to the event loop of the middleman.
virtual void remove_from_loop() = 0; virtual void remove_from_loop() = 0;
......
...@@ -124,6 +124,15 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -124,6 +124,15 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`. /// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd); expected<std::string> remote_addr_of_fd(native_socket fd);
/// Closes the read channel for a socket.
void shutdown_read(native_socket fd);
/// Closes the write channel for a socket.
void shutdown_write(native_socket fd);
/// Closes the both read and write channel for a socket.
void shutdown_both(native_socket fd);
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -35,6 +35,8 @@ public: ...@@ -35,6 +35,8 @@ public:
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void graceful_shutdown() override;
void handle_event(operation op) override; void handle_event(operation op) override;
void init(native_socket sock_fd); void init(native_socket sock_fd);
......
...@@ -43,7 +43,7 @@ public: ...@@ -43,7 +43,7 @@ public:
std::vector<char>& rd_buf() override; std::vector<char>& rd_buf() override;
void stop_reading() override; void graceful_shutdown() override;
void flush() override; void flush() override;
......
...@@ -85,6 +85,8 @@ public: ...@@ -85,6 +85,8 @@ public:
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void graceful_shutdown() override;
/// Forces this stream to subscribe to write events if no data is in the /// Forces this stream to subscribe to write events if no data is in the
/// write buffer. /// write buffer.
void force_empty_write(const manager_ptr& mgr); void force_empty_write(const manager_ptr& mgr);
...@@ -137,6 +139,10 @@ private: ...@@ -137,6 +139,10 @@ private:
void handle_error_propagation(); void handle_error_propagation();
/// Initiates a graceful shutdown of the connection by sending FIN on the TCP
/// connection.
void send_fin();
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
// State for reading. // State for reading.
......
...@@ -18,8 +18,8 @@ ...@@ -18,8 +18,8 @@
#pragma once #pragma once
#include "caf/io/network/rw_state.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/rw_state.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
......
...@@ -340,18 +340,14 @@ bool abstract_broker::remove_endpoint(datagram_handle hdl) { ...@@ -340,18 +340,14 @@ bool abstract_broker::remove_endpoint(datagram_handle hdl) {
void abstract_broker::close_all() { void abstract_broker::close_all() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
while (!doormen_.empty()) { // Calling graceful_shutdown causes the objects to detach from the broker by
// stop_reading will remove the doorman from doormen_ // removing from the container.
doormen_.begin()->second->stop_reading(); while (!doormen_.empty())
} doormen_.begin()->second->graceful_shutdown();
while (!scribes_.empty()) { while (!scribes_.empty())
// stop_reading will remove the scribe from scribes_ scribes_.begin()->second->graceful_shutdown();
scribes_.begin()->second->stop_reading(); while (!datagram_servants_.empty())
} datagram_servants_.begin()->second->graceful_shutdown();
while (!datagram_servants_.empty()) {
// stop reading will remove dgram servants from datagram_servants_
datagram_servants_.begin()->second->stop_reading();
}
} }
resumable::subtype_t abstract_broker::subtype() const { resumable::subtype_t abstract_broker::subtype() const {
......
...@@ -31,7 +31,7 @@ acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd) ...@@ -31,7 +31,7 @@ acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
} }
void acceptor::start(acceptor_manager* mgr) { void acceptor::start(acceptor_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd())); CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
activate(mgr); activate(mgr);
} }
...@@ -43,18 +43,22 @@ void acceptor::activate(acceptor_manager* mgr) { ...@@ -43,18 +43,22 @@ void acceptor::activate(acceptor_manager* mgr) {
} }
} }
void acceptor::stop_reading() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
void acceptor::removed_from_loop(operation op) { void acceptor::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()) << CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG2("fd", fd_) << CAF_ARG(op));
if (op == operation::read) if (op == operation::read)
mgr_.reset(); mgr_.reset();
} }
void acceptor::graceful_shutdown() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Ignore repeated calls.
if (state_.shutting_down)
return;
state_.shutting_down = true;
// Shutdown socket activity.
shutdown_both(fd_);
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -129,6 +129,19 @@ void datagram_handler::removed_from_loop(operation op) { ...@@ -129,6 +129,19 @@ void datagram_handler::removed_from_loop(operation op) {
}; };
} }
void datagram_handler::graceful_shutdown() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Ignore repeated calls.
if (state_.shutting_down)
return;
state_.shutting_down = true;
// Stop reading right away.
passivate();
// UDP is connectionless. Hence, there's no graceful way to shutdown
// anything. This handler gets destroyed automatically once it no longer is
// registered for reading or writing.
}
void datagram_handler::prepare_next_read() { void datagram_handler::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size()) CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size())
<< CAF_ARG(wr_offline_buf_.size())); << CAF_ARG(wr_offline_buf_.size()));
......
...@@ -75,9 +75,9 @@ network::receive_buffer& datagram_servant_impl::rd_buf() { ...@@ -75,9 +75,9 @@ network::receive_buffer& datagram_servant_impl::rd_buf() {
return handler_.rd_buf(); return handler_.rd_buf();
} }
void datagram_servant_impl::stop_reading() { void datagram_servant_impl::graceful_shutdown() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
handler_.stop_reading(); handler_.graceful_shutdown();
detach_handles(); detach_handles();
detach(&handler_.backend(), false); detach(&handler_.backend(), false);
} }
......
...@@ -49,9 +49,9 @@ bool doorman_impl::new_connection() { ...@@ -49,9 +49,9 @@ bool doorman_impl::new_connection() {
return doorman::new_connection(&dm, hdl); return doorman::new_connection(&dm, hdl);
} }
void doorman_impl::stop_reading() { void doorman_impl::graceful_shutdown() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
acceptor_.stop_reading(); acceptor_.graceful_shutdown();
detach(&acceptor_.backend(), false); detach(&acceptor_.backend(), false);
} }
......
...@@ -34,7 +34,7 @@ namespace network { ...@@ -34,7 +34,7 @@ namespace network {
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: fd_(sockfd), : fd_(sockfd),
state_{true, false, false, receive_policy_flag::at_least}, state_{true, false, false, false, receive_policy_flag::at_least},
eventbf_(0), eventbf_(0),
backend_(dm) { backend_(dm) {
set_fd_flags(); set_fd_flags();
...@@ -47,23 +47,10 @@ event_handler::~event_handler() { ...@@ -47,23 +47,10 @@ event_handler::~event_handler() {
} }
} }
void event_handler::close_read_channel() {
if (fd_ == invalid_native_socket || read_channel_closed())
return;
::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX
state_.reading = false;
}
void event_handler::passivate() { void event_handler::passivate() {
backend().del(operation::read, fd(), this); backend().del(operation::read, fd(), this);
} }
void event_handler::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void event_handler::activate() { void event_handler::activate() {
backend().add(operation::read, fd(), this); backend().add(operation::read, fd(), this);
} }
......
...@@ -419,6 +419,34 @@ expected<uint16_t> remote_port_of_fd(native_socket fd) { ...@@ -419,6 +419,34 @@ expected<uint16_t> remote_port_of_fd(native_socket fd) {
return ntohs(port_of(reinterpret_cast<sockaddr&>(st))); return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
} }
// -- shutdown function family -------------------------------------------------
namespace {
#ifdef CAF_WINDOWS
static constexpr int read_channel = SD_RECEIVE;
static constexpr int write_channel = SD_SEND;
static constexpr int both_channels = SD_BOTH;
#else // CAF_WINDOWS
static constexpr int read_channel = SHUT_RD;
static constexpr int write_channel = SHUT_WR;
static constexpr int both_channels = SHUT_RDWR;
#endif // CAF_WINDOWS
} // namespace <anonymous>
void shutdown_read(native_socket fd) {
::shutdown(fd, read_channel);
}
void shutdown_write(native_socket fd) {
::shutdown(fd, write_channel);
}
void shutdown_both(native_socket fd) {
::shutdown(fd, both_channels);
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -43,6 +43,10 @@ void pipe_reader::removed_from_loop(operation) { ...@@ -43,6 +43,10 @@ void pipe_reader::removed_from_loop(operation) {
// nop // nop
} }
void pipe_reader::graceful_shutdown() {
shutdown_read(fd_);
}
resumable* pipe_reader::try_read_next() { resumable* pipe_reader::try_read_next() {
std::intptr_t ptrval; std::intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles // on windows, we actually have sockets, otherwise we have file handles
......
...@@ -55,9 +55,9 @@ std::vector<char>& scribe_impl::rd_buf() { ...@@ -55,9 +55,9 @@ std::vector<char>& scribe_impl::rd_buf() {
return stream_.rd_buf(); return stream_.rd_buf();
} }
void scribe_impl::stop_reading() { void scribe_impl::graceful_shutdown() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.stop_reading(); stream_.graceful_shutdown();
detach(&stream_.backend(), false); detach(&stream_.backend(), false);
} }
......
...@@ -78,7 +78,7 @@ void stream::flush(const manager_ptr& mgr) { ...@@ -78,7 +78,7 @@ void stream::flush(const manager_ptr& mgr) {
} }
void stream::removed_from_loop(operation op) { void stream::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG2("fd", fd_) << CAF_ARG(op));
switch (op) { switch (op) {
case operation::read: reader_.reset(); break; case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break; case operation::write: writer_.reset(); break;
...@@ -86,6 +86,18 @@ void stream::removed_from_loop(operation op) { ...@@ -86,6 +86,18 @@ void stream::removed_from_loop(operation op) {
} }
} }
void stream::graceful_shutdown() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Ignore repeated calls.
if (state_.shutting_down)
return;
state_.shutting_down = true;
// Initiate graceful shutdown unless we have still data to send.
if (!state_.writing)
send_fin();
// Otherwise, send_fin() gets called after draining the send buffer.
}
void stream::force_empty_write(const manager_ptr& mgr) { void stream::force_empty_write(const manager_ptr& mgr) {
if (!state_.writing) { if (!state_.writing) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
...@@ -125,6 +137,8 @@ void stream::prepare_next_write() { ...@@ -125,6 +137,8 @@ void stream::prepare_next_write() {
if (wr_offline_buf_.empty()) { if (wr_offline_buf_.empty()) {
state_.writing = false; state_.writing = false;
backend().del(operation::write, fd(), this); backend().del(operation::write, fd(), this);
if (state_.shutting_down)
send_fin();
} else { } else {
wr_buf_.swap(wr_offline_buf_); wr_buf_.swap(wr_offline_buf_);
} }
...@@ -186,6 +200,15 @@ void stream::handle_error_propagation() { ...@@ -186,6 +200,15 @@ void stream::handle_error_propagation() {
writer_->io_failure(&backend(), operation::write); writer_->io_failure(&backend(), operation::write);
} }
void stream::send_fin() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd_));
// Shutting down the write channel will cause TCP to send FIN for the
// graceful shutdown sequence. The peer then closes its connection as well
// and we will notice this by getting 0 as return value of recv without error
// (connection closed).
shutdown_write(fd_);
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -107,7 +107,7 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) { ...@@ -107,7 +107,7 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) {
std::vector<char>& rd_buf() override { std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl()); return mpx_->input_buffer(hdl());
} }
void stop_reading() override { void graceful_shutdown() override {
mpx_->stopped_reading(hdl()) = true; mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false); detach(mpx_, false);
} }
...@@ -183,7 +183,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) { ...@@ -183,7 +183,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) {
parent()->add_scribe(mpx_->new_scribe(ch)); parent()->add_scribe(mpx_->new_scribe(ch));
return doorman::new_connection(mpx_, ch); return doorman::new_connection(mpx_, ch);
} }
void stop_reading() override { void graceful_shutdown() override {
mpx_->stopped_reading(hdl()) = true; mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false); detach(mpx_, false);
} }
...@@ -364,7 +364,7 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl, ...@@ -364,7 +364,7 @@ datagram_servant_ptr test_multiplexer::new_datagram_servant(datagram_handle hdl,
auto& buf = mpx_->input_buffer(hdl()); auto& buf = mpx_->input_buffer(hdl());
return buf.second; return buf.second;
} }
void stop_reading() override { void graceful_shutdown() override {
mpx_->stopped_reading(hdl()) = true; mpx_->stopped_reading(hdl()) = true;
detach_handles(); detach_handles();
detach(mpx_, false); detach(mpx_, false);
......
...@@ -133,9 +133,9 @@ class scribe_impl : public io::scribe { ...@@ -133,9 +133,9 @@ class scribe_impl : public io::scribe {
return stream_.rd_buf(); return stream_.rd_buf();
} }
void stop_reading() override { void graceful_shutdown() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.stop_reading(); stream_.graceful_shutdown();
detach(&stream_.backend(), false); detach(&stream_.backend(), false);
} }
......
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