Commit e3591562 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/735'

parents b6a31c78 3e12e212
...@@ -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) {
......
...@@ -57,8 +57,6 @@ public: ...@@ -57,8 +57,6 @@ public:
/// Activates the datagram handler. /// Activates the datagram handler.
void activate(datagram_manager* mgr); void activate(datagram_manager* mgr);
void ack_writes(bool x);
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @warning Not thread safe. /// @warning Not thread safe.
void write(datagram_handle hdl, const void* buf, size_t num_bytes); void write(datagram_handle hdl, const void* buf, size_t num_bytes);
...@@ -66,7 +64,7 @@ public: ...@@ -66,7 +64,7 @@ public:
/// Returns the write buffer of this enpoint. /// Returns the write buffer of this enpoint.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline write_buffer_type& wr_buf(datagram_handle hdl) { write_buffer_type& wr_buf(datagram_handle hdl) {
wr_offline_buf_.emplace_back(); wr_offline_buf_.emplace_back();
wr_offline_buf_.back().first = hdl; wr_offline_buf_.back().first = hdl;
return wr_offline_buf_.back().second; return wr_offline_buf_.back().second;
...@@ -75,14 +73,14 @@ public: ...@@ -75,14 +73,14 @@ public:
/// Enqueues a buffer to be sent as a datagram. /// Enqueues a buffer to be sent as a datagram.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) { void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) {
wr_offline_buf_.emplace_back(hdl, move(buf)); wr_offline_buf_.emplace_back(hdl, move(buf));
} }
/// Returns the read buffer of this stream. /// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline read_buffer_type& rd_buf() { read_buffer_type& rd_buf() {
return rd_buf_; return rd_buf_;
} }
...@@ -92,12 +90,10 @@ public: ...@@ -92,12 +90,10 @@ public:
/// once the stream has been started. /// once the stream has been started.
void flush(const manager_ptr& mgr); void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket 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;
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);
...@@ -107,7 +103,7 @@ public: ...@@ -107,7 +103,7 @@ public:
void remove_endpoint(datagram_handle hdl); void remove_endpoint(datagram_handle hdl);
inline ip_endpoint& sending_endpoint() { ip_endpoint& sending_endpoint() {
return sender_; return sender_;
} }
...@@ -180,8 +176,6 @@ private: ...@@ -180,8 +176,6 @@ private:
// state for writing // state for writing
int send_buffer_size_; int send_buffer_size_;
bool ack_writes_;
bool writing_;
std::deque<job_type> wr_offline_buf_; std::deque<job_type> wr_offline_buf_;
job_type wr_buf_; job_type wr_buf_;
manager_ptr writer_; manager_ptr writer_;
......
...@@ -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;
......
...@@ -19,9 +19,9 @@ ...@@ -19,9 +19,9 @@
#pragma once #pragma once
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/receive_policy.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -30,6 +30,24 @@ namespace network { ...@@ -30,6 +30,24 @@ 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 {
/// Stores whether the socket is currently registered for reading.
bool reading : 1;
/// Stores whether the socket is currently registered for writing.
bool writing : 1;
/// Stores whether the parent actor demanded write receipts.
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;
};
event_handler(default_multiplexer& dm, native_socket sockfd); event_handler(default_multiplexer& dm, native_socket sockfd);
virtual ~event_handler(); virtual ~event_handler();
...@@ -43,46 +61,60 @@ public: ...@@ -43,46 +61,60 @@ 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.
inline native_socket fd() const { native_socket fd() const {
return fd_; return fd_;
} }
/// Returns the `multiplexer` this acceptor belongs to. /// Returns the `multiplexer` this acceptor belongs to.
inline default_multiplexer& backend() { default_multiplexer& backend() {
return backend_; return backend_;
} }
/// Returns the bit field storing the subscribed events. /// Returns the bit field storing the subscribed events.
inline int eventbf() const { int eventbf() const {
return eventbf_; return eventbf_;
} }
/// Sets the bit field storing the subscribed events. /// Sets the bit field storing the subscribed events.
inline void eventbf(int value) { void eventbf(int value) {
eventbf_ = value; eventbf_ = value;
} }
/// Checks whether `close_read` has been called. /// Checks whether `close_read_channel` has been called.
inline bool read_channel_closed() const { bool read_channel_closed() const {
return read_channel_closed_; 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();
/// Returns whether this event handlers signals successful writes to its
/// parent actor.
bool ack_writes() {
return state_.ack_writes;
}
/// Sets whether this event handlers signals successful writes to its parent
/// actor.
void ack_writes(bool x) {
state_.ack_writes = x;
}
protected: protected:
/// Adds the file descriptor to the event loop of the parent. /// Adds the file descriptor to the event loop of the parent.
void activate(); void activate();
/// Sets flags for asynchronous event handling on the socket handle.
void set_fd_flags(); void set_fd_flags();
int eventbf_;
native_socket fd_; native_socket fd_;
bool read_channel_closed_; state state_;
int eventbf_;
default_multiplexer& backend_; default_multiplexer& backend_;
}; };
......
...@@ -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;
......
...@@ -59,8 +59,6 @@ public: ...@@ -59,8 +59,6 @@ public:
/// once the stream has been started. /// once the stream has been started.
void configure_read(receive_policy::config config); void configure_read(receive_policy::config config);
void ack_writes(bool x);
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @warning Not thread safe. /// @warning Not thread safe.
void write(const void* buf, size_t num_bytes); void write(const void* buf, size_t num_bytes);
...@@ -85,12 +83,10 @@ public: ...@@ -85,12 +83,10 @@ public:
/// once the stream has been started. /// once the stream has been started.
void flush(const manager_ptr& mgr); void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket 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;
/// 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);
...@@ -143,6 +139,10 @@ private: ...@@ -143,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.
...@@ -150,13 +150,10 @@ private: ...@@ -150,13 +150,10 @@ private:
size_t read_threshold_; size_t read_threshold_;
size_t collected_; size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_;
buffer_type rd_buf_; buffer_type rd_buf_;
// State for writing. // State for writing.
manager_ptr writer_; manager_ptr writer_;
bool ack_writes_;
bool writing_;
size_t written_; size_t written_;
buffer_type wr_buf_; buffer_type wr_buf_;
buffer_type wr_offline_buf_; buffer_type wr_offline_buf_;
......
...@@ -27,7 +27,7 @@ ...@@ -27,7 +27,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
enum class receive_policy_flag { enum class receive_policy_flag : unsigned {
at_least, at_least,
at_most, at_most,
exactly exactly
......
...@@ -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
...@@ -44,9 +44,7 @@ datagram_handler::datagram_handler(default_multiplexer& backend_ref, ...@@ -44,9 +44,7 @@ datagram_handler::datagram_handler(default_multiplexer& backend_ref,
defaults::middleman::max_consecutive_reads)), defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size), max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size), rd_buf_(receive_buffer_size),
send_buffer_size_(0), send_buffer_size_(0) {
ack_writes_(false),
writing_(false) {
allow_udp_connreset(sockfd, false); allow_udp_connreset(sockfd, false);
auto es = send_buffer_size(sockfd); auto es = send_buffer_size(sockfd);
if (!es) if (!es)
...@@ -69,10 +67,6 @@ void datagram_handler::activate(datagram_manager* mgr) { ...@@ -69,10 +67,6 @@ void datagram_handler::activate(datagram_manager* mgr) {
} }
} }
void datagram_handler::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_handler::write(datagram_handle hdl, const void* buf, void datagram_handler::write(datagram_handle hdl, const void* buf,
size_t num_bytes) { size_t num_bytes) {
wr_offline_buf_.emplace_back(); wr_offline_buf_.emplace_back();
...@@ -85,10 +79,10 @@ void datagram_handler::write(datagram_handle hdl, const void* buf, ...@@ -85,10 +79,10 @@ void datagram_handler::write(datagram_handle hdl, const void* buf,
void datagram_handler::flush(const manager_ptr& mgr) { void datagram_handler::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) { if (!wr_offline_buf_.empty() && !state_.writing) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
writer_ = mgr; writer_ = mgr;
writing_ = true; state_.writing = true;
prepare_next_write(); prepare_next_write();
} }
} }
...@@ -127,12 +121,6 @@ void datagram_handler::remove_endpoint(datagram_handle hdl) { ...@@ -127,12 +121,6 @@ void datagram_handler::remove_endpoint(datagram_handle hdl) {
} }
} }
void datagram_handler::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_handler::removed_from_loop(operation op) { void datagram_handler::removed_from_loop(operation op) {
switch (op) { switch (op) {
case operation::read: reader_.reset(); break; case operation::read: reader_.reset(); break;
...@@ -141,6 +129,19 @@ void datagram_handler::removed_from_loop(operation op) { ...@@ -141,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()));
...@@ -151,7 +152,7 @@ void datagram_handler::prepare_next_write() { ...@@ -151,7 +152,7 @@ void datagram_handler::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
wr_buf_.second.clear(); wr_buf_.second.clear();
if (wr_offline_buf_.empty()) { if (wr_offline_buf_.empty()) {
writing_ = false; state_.writing = false;
backend().del(operation::write, fd(), this); backend().del(operation::write, fd(), this);
} else { } else {
wr_buf_.swap(wr_offline_buf_.front()); wr_buf_.swap(wr_offline_buf_.front());
...@@ -189,7 +190,7 @@ void datagram_handler::handle_write_result(bool write_result, datagram_handle id ...@@ -189,7 +190,7 @@ void datagram_handler::handle_write_result(bool write_result, datagram_handle id
backend().del(operation::write, fd(), this); backend().del(operation::write, fd(), this);
} else if (wb > 0) { } else if (wb > 0) {
CAF_ASSERT(wb == buf.size()); CAF_ASSERT(wb == buf.size());
if (ack_writes_) if (state_.ack_writes)
writer_->datagram_sent(&backend(), id, wb, std::move(buf)); writer_->datagram_sent(&backend(), id, wb, std::move(buf));
prepare_next_write(); prepare_next_write();
} else { } else {
......
...@@ -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);
} }
......
...@@ -33,9 +33,9 @@ namespace io { ...@@ -33,9 +33,9 @@ namespace io {
namespace network { namespace network {
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0), : fd_(sockfd),
fd_(sockfd), state_{true, false, false, false, receive_policy_flag::at_least},
read_channel_closed_(false), eventbf_(0),
backend_(dm) { backend_(dm) {
set_fd_flags(); set_fd_flags();
} }
...@@ -47,13 +47,6 @@ event_handler::~event_handler() { ...@@ -47,13 +47,6 @@ 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
read_channel_closed_ = true;
}
void event_handler::passivate() { void event_handler::passivate() {
backend().del(operation::read, fd(), this); backend().del(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);
} }
......
...@@ -37,8 +37,6 @@ stream::stream(default_multiplexer& backend_ref, native_socket sockfd) ...@@ -37,8 +37,6 @@ stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
defaults::middleman::max_consecutive_reads)), defaults::middleman::max_consecutive_reads)),
read_threshold_(1), read_threshold_(1),
collected_(0), collected_(0),
ack_writes_(false),
writing_(false),
written_(0) { written_(0) {
configure_read(receive_policy::at_most(1024)); configure_read(receive_policy::at_most(1024));
} }
...@@ -57,14 +55,10 @@ void stream::activate(stream_manager* mgr) { ...@@ -57,14 +55,10 @@ void stream::activate(stream_manager* mgr) {
} }
void stream::configure_read(receive_policy::config config) { void stream::configure_read(receive_policy::config config) {
rd_flag_ = config.first; state_.rd_flag = config.first;
max_ = config.second; max_ = config.second;
} }
void stream::ack_writes(bool x) {
ack_writes_ = x;
}
void stream::write(const void* buf, size_t num_bytes) { void stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes)); CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf); auto first = reinterpret_cast<const char*>(buf);
...@@ -75,22 +69,16 @@ void stream::write(const void* buf, size_t num_bytes) { ...@@ -75,22 +69,16 @@ void stream::write(const void* buf, size_t num_bytes) {
void stream::flush(const manager_ptr& mgr) { void stream::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) { if (!wr_offline_buf_.empty() && !state_.writing) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
writer_ = mgr; writer_ = mgr;
writing_ = true; state_.writing = true;
prepare_next_write(); prepare_next_write();
} }
} }
void stream::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
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;
...@@ -98,17 +86,31 @@ void stream::removed_from_loop(operation op) { ...@@ -98,17 +86,31 @@ 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 (!writing_) { if (!state_.writing) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
writer_ = mgr; writer_ = mgr;
writing_ = true; state_.writing = true;
} }
} }
void stream::prepare_next_read() { void stream::prepare_next_read() {
collected_ = 0; collected_ = 0;
switch (rd_flag_) { // This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<receive_policy_flag>(state_.rd_flag)) {
case receive_policy_flag::exactly: case receive_policy_flag::exactly:
if (rd_buf_.size() != max_) if (rd_buf_.size() != max_)
rd_buf_.resize(max_); rd_buf_.resize(max_);
...@@ -135,8 +137,10 @@ void stream::prepare_next_write() { ...@@ -135,8 +137,10 @@ void stream::prepare_next_write() {
written_ = 0; written_ = 0;
wr_buf_.clear(); wr_buf_.clear();
if (wr_offline_buf_.empty()) { if (wr_offline_buf_.empty()) {
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_);
} }
...@@ -181,7 +185,7 @@ void stream::handle_write_result(rw_state write_result, size_t wb) { ...@@ -181,7 +185,7 @@ void stream::handle_write_result(rw_state write_result, size_t wb) {
written_ += wb; written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size()); CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_; auto remaining = wr_buf_.size() - written_;
if (ack_writes_) if (state_.ack_writes)
writer_->data_transferred(&backend(), wb, writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size()); remaining + wr_offline_buf_.size());
// prepare next send (or stop sending) // prepare next send (or stop sending)
...@@ -198,6 +202,15 @@ void stream::handle_error_propagation() { ...@@ -198,6 +202,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