Commit e5057c0a authored by Dominik Charousset's avatar Dominik Charousset

Consolidate event handler implementation

- Collect various flags in one bitfield
- Move identical functions in stream and datagram_handle into base type
parent 9cdc4b4b
...@@ -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,10 +90,6 @@ public: ...@@ -92,10 +90,6 @@ 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 add_endpoint(datagram_handle hdl, const ip_endpoint& ep, void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
...@@ -107,7 +101,7 @@ public: ...@@ -107,7 +101,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 +174,6 @@ private: ...@@ -180,8 +174,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_;
......
...@@ -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,13 @@ namespace network { ...@@ -30,6 +30,13 @@ namespace network {
/// A socket I/O event handler. /// A socket I/O event handler.
class event_handler { class event_handler {
public: public:
struct state {
bool reading : 1;
bool writing : 1;
bool ack_writes : 1;
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();
...@@ -44,28 +51,28 @@ public: ...@@ -44,28 +51,28 @@ public:
virtual void removed_from_loop(operation op) = 0; virtual void removed_from_loop(operation op) = 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. /// Closes the read channel of the underlying socket.
...@@ -74,15 +81,32 @@ public: ...@@ -74,15 +81,32 @@ public:
/// 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
/// 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_;
}; };
......
...@@ -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,10 +83,6 @@ public: ...@@ -85,10 +83,6 @@ 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;
/// 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
...@@ -150,13 +144,10 @@ private: ...@@ -150,13 +144,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
......
...@@ -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;
...@@ -151,7 +139,7 @@ void datagram_handler::prepare_next_write() { ...@@ -151,7 +139,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 +177,7 @@ void datagram_handler::handle_write_result(bool write_result, datagram_handle id ...@@ -189,7 +177,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 {
......
...@@ -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, receive_policy_flag::at_least},
read_channel_closed_(false), eventbf_(0),
backend_(dm) { backend_(dm) {
set_fd_flags(); set_fd_flags();
} }
...@@ -48,16 +48,22 @@ event_handler::~event_handler() { ...@@ -48,16 +48,22 @@ event_handler::~event_handler() {
} }
void event_handler::close_read_channel() { void event_handler::close_read_channel() {
if (fd_ == invalid_native_socket || read_channel_closed_) if (fd_ == invalid_native_socket || read_channel_closed())
return; return;
::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX ::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX
read_channel_closed_ = true; 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);
} }
......
...@@ -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,20 +69,14 @@ void stream::write(const void* buf, size_t num_bytes) { ...@@ -75,20 +69,14 @@ 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_ARG(op));
switch (op) { switch (op) {
...@@ -99,16 +87,16 @@ void stream::removed_from_loop(operation op) { ...@@ -99,16 +87,16 @@ void stream::removed_from_loop(operation op) {
} }
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_) { switch (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,7 +123,7 @@ void stream::prepare_next_write() { ...@@ -135,7 +123,7 @@ 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);
} else { } else {
wr_buf_.swap(wr_offline_buf_); wr_buf_.swap(wr_offline_buf_);
...@@ -181,7 +169,7 @@ void stream::handle_write_result(rw_state write_result, size_t wb) { ...@@ -181,7 +169,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)
......
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