Commit f1cac1c5 authored by Dominik Charousset's avatar Dominik Charousset

Remove socket_had_rd_shutdown_event

parent b82ff29c
...@@ -111,7 +111,7 @@ namespace network { ...@@ -111,7 +111,7 @@ namespace network {
constexpr int ec_interrupted_syscall = EINTR; constexpr int ec_interrupted_syscall = EINTR;
#endif #endif
// poll xs epoll backend // poll vs epoll backend
#if ! defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer #if ! defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
// From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft // From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft
...@@ -181,10 +181,10 @@ bool try_accept(native_socket& result, native_socket fd); ...@@ -181,10 +181,10 @@ bool try_accept(native_socket& result, native_socket fd);
class default_multiplexer; class default_multiplexer;
/// A socket IO event handler. /// A socket I/O event handler.
class event_handler { class event_handler {
public: public:
event_handler(default_multiplexer& dm); event_handler(default_multiplexer& dm, native_socket fd);
virtual ~event_handler(); virtual ~event_handler();
...@@ -198,7 +198,9 @@ public: ...@@ -198,7 +198,9 @@ 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.
virtual native_socket fd() const = 0; inline native_socket fd() const {
return fd_;
}
/// Returns the `multiplexer` this acceptor belongs to. /// Returns the `multiplexer` this acceptor belongs to.
inline default_multiplexer& backend() { inline default_multiplexer& backend() {
...@@ -215,46 +217,32 @@ public: ...@@ -215,46 +217,32 @@ public:
eventbf_ = value; eventbf_ = value;
} }
protected: /// Checks whether `close_read` has been called.
default_multiplexer& backend_; inline bool read_channel_closed() const {
int eventbf_; return read_channel_closed_;
};
/// Low-level socket type used as default.
class default_socket {
public:
using socket_type = default_socket;
default_socket(default_multiplexer& parent,
native_socket sock = invalid_native_socket);
default_socket(default_socket&& other);
default_socket& operator=(default_socket&& other);
~default_socket();
void close_read();
inline native_socket fd() const {
return fd_;
} }
inline native_socket native_handle() const { /// Closes the read channel of the underlying socket.
return fd_; void close_read_channel();
}
inline default_multiplexer& backend() { protected:
return parent_; void set_fd_flags();
}
private: int eventbf_;
default_multiplexer& parent_;
native_socket fd_; native_socket fd_;
bool read_channel_closed_;
default_multiplexer& backend_;
}; };
/// Low-level socket type used as default. /// An event handler for the internal event pipe.
using default_socket_acceptor = default_socket; class pipe_reader : public event_handler {
public:
pipe_reader(default_multiplexer& dm);
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
void init(native_socket fd);
resumable* try_read_next();
};
class default_multiplexer : public multiplexer { class default_multiplexer : public multiplexer {
public: public:
...@@ -283,8 +271,6 @@ public: ...@@ -283,8 +271,6 @@ public:
void assign_tcp_scribe(abstract_broker* ptr, connection_handle hdl) override; void assign_tcp_scribe(abstract_broker* ptr, connection_handle hdl) override;
connection_handle add_tcp_scribe(abstract_broker*, default_socket_acceptor&&);
connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override; connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override;
connection_handle add_tcp_scribe(abstract_broker*, const std::string& h, connection_handle add_tcp_scribe(abstract_broker*, const std::string& h,
...@@ -295,8 +281,6 @@ public: ...@@ -295,8 +281,6 @@ public:
void assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) override; void assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override; accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
...@@ -361,39 +345,32 @@ private: ...@@ -361,39 +345,32 @@ private:
void handle(const event& event); void handle(const event& event);
bool socket_had_rd_shutdown_event(native_socket fd);
void handle_socket_event(native_socket fd, int mask, event_handler* ptr); void handle_socket_event(native_socket fd, int mask, event_handler* ptr);
void close_pipe(); void close_pipe();
void wr_dispatch_request(resumable* ptr); void wr_dispatch_request(resumable* ptr);
resumable* rd_dispatch_request(); //resumable* rd_dispatch_request();
native_socket epollfd_; // unused in poll() implementation native_socket epollfd_; // unused in poll() implementation
std::vector<multiplexer_data> pollset_; std::vector<multiplexer_data> pollset_;
std::vector<event> events_; // always sorted by .fd std::vector<event> events_; // always sorted by .fd
multiplexer_poll_shadow_data shadow_; multiplexer_poll_shadow_data shadow_;
std::pair<native_socket, native_socket> pipe_; std::pair<native_socket, native_socket> pipe_;
pipe_reader pipe_reader_;
}; };
template <class T> inline connection_handle conn_hdl_from_socket(native_socket fd) {
connection_handle conn_hdl_from_socket(const T& sock) { return connection_handle::from_int(int64_from_native_socket(fd));
return connection_handle::from_int(
int64_from_native_socket(sock.native_handle()));
} }
template <class T> inline accept_handle accept_hdl_from_socket(native_socket fd) {
accept_handle accept_hdl_from_socket(const T& sock) { return accept_handle::from_int(int64_from_native_socket(fd));
return accept_handle::from_int(
int64_from_native_socket(sock.native_handle()));
} }
/// A stream capable of both reading and writing. The stream's input /// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}. /// data is forwarded to its {@link stream_manager manager}.
template <class Socket>
class stream : public event_handler { class stream : public event_handler {
public: public:
/// A smart pointer to a stream manager. /// A smart pointer to a stream manager.
...@@ -403,72 +380,33 @@ public: ...@@ -403,72 +380,33 @@ public:
/// interface to `std::vector`. /// interface to `std::vector`.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
stream(default_multiplexer& backend_ref) stream(default_multiplexer& backend_ref, native_socket sockfd);
: event_handler(backend_ref),
sock_(backend_ref),
read_threshold_(1),
collected_(0),
ack_writes_(false),
writing_(false),
written_(0) {
configure_read(receive_policy::at_most(1024));
}
/// Returns the IO socket.
Socket& socket_handle() {
return sock_;
}
/// Initializes this stream, setting the socket handle to `fd`.
void init(Socket sockfd) {
sock_ = std::move(sockfd);
}
/// Starts reading data from the socket, forwarding incoming data to `mgr`. /// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(const manager_ptr& mgr) { void start(const manager_ptr& mgr);
CAF_ASSERT(mgr != nullptr);
reader_ = mgr;
backend().add(operation::read, sock_.fd(), this);
read_loop();
}
void removed_from_loop(operation op) override {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
}
}
/// Configures how much data will be provided for the next `consume` callback. /// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
/// 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);
rd_flag_ = config.first;
max_ = config.second;
}
void ack_writes(bool x) { void ack_writes(bool x);
ack_writes_ = 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);
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
/// Returns the write buffer of this stream. /// Returns the write 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.
buffer_type& wr_buf() { inline buffer_type& wr_buf() {
return wr_offline_buf_; return wr_offline_buf_;
} }
buffer_type& rd_buf() { /// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_; return rd_buf_;
} }
...@@ -476,128 +414,30 @@ public: ...@@ -476,128 +414,30 @@ public:
/// member function of `mgr` in case of an error. /// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
void flush(const manager_ptr& mgr) { void flush(const manager_ptr& mgr);
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (! wr_offline_buf_.empty() && ! writing_) {
backend().add(operation::write, sock_.fd(), this);
writer_ = mgr;
writing_ = true;
write_loop();
}
}
void stop_reading() { /// Closes the read channel of the underlying socket and removes
CAF_LOG_TRACE(""); /// this handler from its parent.
sock_.close_read(); void stop_reading();
backend().del(operation::read, sock_.fd(), this);
}
void handle_event(operation op) override { void removed_from_loop(operation op) override;
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
size_t rb; // read bytes
if (! read_some(rb, sock_.fd(),
rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
backend().del(operation::read, sock_.fd(), this);
} else if (rb > 0) {
collected_ += rb;
if (collected_ >= read_threshold_) {
reader_->consume(&backend(), rd_buf_.data(), collected_);
read_loop();
}
}
break;
}
case operation::write: {
size_t wb; // written bytes
if (! write_some(wb, sock_.fd(),
wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, sock_.fd(), this);
} else if (wb > 0) {
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
write_loop();
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
}
native_socket fd() const override { void handle_event(operation op) override;
return sock_.fd();
}
private: private:
void read_loop() { void read_loop();
collected_ = 0;
switch (rd_flag_) {
case receive_policy_flag::exactly:
if (rd_buf_.size() != max_) {
rd_buf_.resize(max_);
}
read_threshold_ = max_;
break;
case receive_policy_flag::at_most:
if (rd_buf_.size() != max_) {
rd_buf_.resize(max_);
}
read_threshold_ = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_
+ std::max<size_t>(100, max_ / 10);
if (rd_buf_.size() != max_size) {
rd_buf_.resize(max_size);
}
read_threshold_ = max_;
break;
}
}
}
void write_loop() { void write_loop();
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
written_ = 0;
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, sock_.fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
}
}
// reading & writing // state for reading
Socket sock_;
// reading
manager_ptr reader_; manager_ptr reader_;
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_; receive_policy_flag rd_flag_;
buffer_type rd_buf_; buffer_type rd_buf_;
// writing
// state for writing
manager_ptr writer_; manager_ptr writer_;
bool ack_writes_; bool ack_writes_;
bool writing_; bool writing_;
...@@ -607,101 +447,45 @@ private: ...@@ -607,101 +447,45 @@ private:
}; };
/// An acceptor is responsible for accepting incoming connections. /// An acceptor is responsible for accepting incoming connections.
template <class SocketAcceptor>
class acceptor : public event_handler { class acceptor : public event_handler {
public: public:
using socket_type = typename SocketAcceptor::socket_type;
/// A manager providing the `accept` member function. /// A manager providing the `accept` member function.
using manager_type = acceptor_manager; using manager_type = acceptor_manager;
/// A smart pointer to an acceptor manager. /// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>; using manager_ptr = intrusive_ptr<manager_type>;
acceptor(default_multiplexer& backend_ref) acceptor(default_multiplexer& backend_ref, native_socket sockfd);
: event_handler(backend_ref),
accept_sock_(backend_ref),
sock_(backend_ref) {
// nop
}
/// Returns the IO socket.
SocketAcceptor& socket_handle() {
return accept_sock_;
}
/// Returns the accepted socket. This member function should /// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback. /// be called only from the `new_connection` callback.
socket_type& accepted_socket() { inline native_socket& accepted_socket() {
return sock_; return sock_;
} }
/// Initializes this acceptor, setting the socket handle to `fd`.
void init(SocketAcceptor sock) {
CAF_LOG_TRACE(CAF_ARG(sock.fd()));
accept_sock_ = std::move(sock);
}
/// Starts this acceptor, forwarding all incoming connections to /// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the /// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured. /// acceptor has been closed or an IO error occured.
void start(const manager_ptr& mgr) { void start(const manager_ptr& mgr);
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()));
CAF_ASSERT(mgr != nullptr);
mgr_ = mgr;
backend().add(operation::read, accept_sock_.fd(), this);
}
/// Closes the network connection, thus stopping this acceptor. /// Closes the network connection and removes this handler from its parent.
void stop_reading() { void stop_reading();
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()));
backend().del(operation::read, accept_sock_.fd(), this);
accept_sock_.close_read();
}
void handle_event(operation op) override {
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (try_accept(sockfd, accept_sock_.fd())) {
if (sockfd != invalid_native_socket) {
sock_ = socket_type{backend(), sockfd};
mgr_->new_connection();
}
}
}
}
void removed_from_loop(operation op) override { void handle_event(operation op) override;
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()) << CAF_ARG(op));
if (op == operation::read)
mgr_.reset();
}
native_socket fd() const override { void removed_from_loop(operation op) override;
return accept_sock_.fd();
}
private: private:
manager_ptr mgr_; manager_ptr mgr_;
SocketAcceptor accept_sock_; native_socket sock_;
socket_type sock_;
}; };
native_socket native_socket new_tcp_connection(const std::string& host, uint16_t port,
new_tcp_connection_impl(const std::string&, uint16_t, maybe<protocol> = none); maybe<protocol> preferred = none);
default_socket
new_tcp_connection(default_multiplexer& backend,
const std::string& host, uint16_t port);
std::pair<native_socket, uint16_t> std::pair<native_socket, uint16_t>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr); new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
std::pair<default_socket_acceptor, uint16_t>
new_tcp_acceptor(default_multiplexer& backend, uint16_t port,
const char* addr = nullptr, bool reuse_addr = false);
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
# include <winsock2.h> # include <winsock2.h>
# include <ws2tcpip.h> /* socklen_t, et al (MSVC20xx) */ # include <ws2tcpip.h> // socklen_t, etc. (MSVC20xx)
# include <windows.h> # include <windows.h>
# include <io.h> # include <io.h>
#else #else
...@@ -270,7 +270,7 @@ uint16_t remote_port_of_fd(native_socket fd); ...@@ -270,7 +270,7 @@ uint16_t remote_port_of_fd(native_socket fd);
#endif #endif
/****************************************************************************** /******************************************************************************
* epoll() xs. poll() * * epoll() vs. poll() *
******************************************************************************/ ******************************************************************************/
#ifdef CAF_EPOLL_MULTIPLEXER #ifdef CAF_EPOLL_MULTIPLEXER
...@@ -281,7 +281,8 @@ uint16_t remote_port_of_fd(native_socket fd); ...@@ -281,7 +281,8 @@ uint16_t remote_port_of_fd(native_socket fd);
default_multiplexer::default_multiplexer(actor_system* sys) default_multiplexer::default_multiplexer(actor_system* sys)
: multiplexer(sys), : multiplexer(sys),
epollfd_(invalid_native_socket), epollfd_(invalid_native_socket),
shadow_(1) { shadow_(1),
pipe_reader_(*this) {
init(); init();
epollfd_ = epoll_create1(EPOLL_CLOEXEC); epollfd_ = epoll_create1(EPOLL_CLOEXEC);
if (epollfd_ == -1) { if (epollfd_ == -1) {
...@@ -291,10 +292,11 @@ uint16_t remote_port_of_fd(native_socket fd); ...@@ -291,10 +292,11 @@ uint16_t remote_port_of_fd(native_socket fd);
// handle at most 64 events at a time // handle at most 64 events at a time
pollset_.resize(64); pollset_.resize(64);
pipe_ = create_pipe(); pipe_ = create_pipe();
pipe_reader.init(pipe_.first);
epoll_event ee; epoll_event ee;
ee.events = input_mask; ee.events = input_mask;
ee.data.ptr = nullptr; ee.data.ptr = &pipe_reader;
if (epoll_ctl(epollfd_, EPOLL_CTL_ADD, pipe_.first, &ee) < 0) { if (epoll_ctl(epollfd_, EPOLL_CTL_ADD, pipe_reader.fd(), &ee) < 0) {
CAF_LOG_ERROR("epoll_ctl: " << strerror(errno)); CAF_LOG_ERROR("epoll_ctl: " << strerror(errno));
exit(errno); exit(errno);
} }
...@@ -417,16 +419,18 @@ uint16_t remote_port_of_fd(native_socket fd); ...@@ -417,16 +419,18 @@ uint16_t remote_port_of_fd(native_socket fd);
default_multiplexer::default_multiplexer(actor_system* sys) default_multiplexer::default_multiplexer(actor_system* sys)
: multiplexer(sys), : multiplexer(sys),
epollfd_(-1) { epollfd_(-1),
pipe_reader_(*this) {
init(); init();
// initial setup // initial setup
pipe_ = create_pipe(); pipe_ = create_pipe();
pipe_reader_.init(pipe_.first);
pollfd pipefd; pollfd pipefd;
pipefd.fd = pipe_.first; pipefd.fd = pipe_reader_.fd();
pipefd.events = input_mask; pipefd.events = input_mask;
pipefd.revents = 0; pipefd.revents = 0;
pollset_.push_back(pipefd); pollset_.push_back(pipefd);
shadow_.push_back(nullptr); shadow_.push_back(&pipe_reader_);
} }
void default_multiplexer::run() { void default_multiplexer::run() {
...@@ -626,21 +630,6 @@ void default_multiplexer::wr_dispatch_request(resumable* ptr) { ...@@ -626,21 +630,6 @@ void default_multiplexer::wr_dispatch_request(resumable* ptr) {
} }
} }
resumable* default_multiplexer::rd_dispatch_request() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(pipe_.first, reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto res = read(pipe_.first, &ptrval, sizeof(ptrval));
# endif
if (res != sizeof(ptrval)) {
return nullptr;
}
return reinterpret_cast<resumable*>(ptrval);;
}
multiplexer::supervisor_ptr default_multiplexer::make_supervisor() { multiplexer::supervisor_ptr default_multiplexer::make_supervisor() {
class impl : public multiplexer::supervisor { class impl : public multiplexer::supervisor {
public: public:
...@@ -662,63 +651,29 @@ void default_multiplexer::close_pipe() { ...@@ -662,63 +651,29 @@ void default_multiplexer::close_pipe() {
del(operation::read, pipe_.first, nullptr); del(operation::read, pipe_.first, nullptr);
} }
bool default_multiplexer::socket_had_rd_shutdown_event(native_socket fd) {
auto last = events_.end();
auto i = std::lower_bound(events_.begin(), last, fd, event_less{});
if (i != last && i->fd == fd) {
// socket is about to be shut down for read if
// its new bitmask does not have the input_mask flag
return (i->mask & input_mask) == 0;
}
return false;
}
void default_multiplexer::handle_socket_event(native_socket fd, int mask, void default_multiplexer::handle_socket_event(native_socket fd, int mask,
event_handler* ptr) { event_handler* ptr) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask));
CAF_ASSERT(ptr != nullptr);
bool checkerror = true; bool checkerror = true;
if (mask & input_mask) {
checkerror = false;
// ignore read events if a previous event caused // ignore read events if a previous event caused
// this socket to be shut down for reading // this socket to be shut down for reading
if ((mask & input_mask) && ! socket_had_rd_shutdown_event(fd)) { if (! ptr->read_channel_closed())
checkerror = false;
if (ptr) {
ptr->handle_event(operation::read); ptr->handle_event(operation::read);
} else {
CAF_ASSERT(fd == pipe_.first);
CAF_LOG_DEBUG("read message from pipe");
auto cb = rd_dispatch_request();
switch (cb->resume(this, max_throughput())) {
case resumable::resume_later:
exec_later(cb);
break;
case resumable::done:
intrusive_ptr_release(cb);
break;
default:
; // ignored
}
}
} }
if (mask & output_mask) { if (mask & output_mask) {
// we do *never* register our pipe handle for writing
CAF_ASSERT(ptr != nullptr);
checkerror = false; checkerror = false;
ptr->handle_event(operation::write); ptr->handle_event(operation::write);
} }
if (checkerror && (mask & error_mask)) { if (checkerror && (mask & error_mask)) {
if (ptr) {
CAF_LOG_DEBUG("error occured on socket:" CAF_LOG_DEBUG("error occured on socket:"
<< CAF_ARG(fd) << CAF_ARG(last_socket_error()) << CAF_ARG(fd) << CAF_ARG(last_socket_error())
<< CAF_ARG(last_socket_error_as_string())); << CAF_ARG(last_socket_error_as_string()));
ptr->handle_event(operation::propagate_error); ptr->handle_event(operation::propagate_error);
del(operation::read, fd, ptr); del(operation::read, fd, ptr);
del(operation::write, fd, ptr); del(operation::write, fd, ptr);
} else {
CAF_LOG_DEBUG("pipe has been closed, assume shutdown");
// a pipe failure means we are in the process
// of shutting down the middleman
del(operation::read, fd, nullptr);
}
} }
} }
...@@ -732,24 +687,26 @@ void default_multiplexer::init() { ...@@ -732,24 +687,26 @@ void default_multiplexer::init() {
} }
default_multiplexer::~default_multiplexer() { default_multiplexer::~default_multiplexer() {
if (epollfd_ != invalid_native_socket) { if (epollfd_ != invalid_native_socket)
closesocket(epollfd_); closesocket(epollfd_);
}
// close write handle first // close write handle first
closesocket(pipe_.second); closesocket(pipe_.second);
// flush pipe before closing it // flush pipe before closing it
nonblocking(pipe_.first, true); nonblocking(pipe_.first, true);
auto ptr = rd_dispatch_request(); auto ptr = pipe_reader_.try_read_next();
while (ptr) { while (ptr) {
intrusive_ptr_release(ptr); intrusive_ptr_release(ptr);
ptr = rd_dispatch_request(); ptr = pipe_reader_.try_read_next();
} }
closesocket(pipe_.first); // do cleanup for pipe reader manually, since WSACleanup needs to happen last
closesocket(pipe_reader_.fd());
pipe_reader_.init(invalid_native_socket);
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
WSACleanup(); WSACleanup();
# endif # endif
} }
void default_multiplexer::exec_later(resumable* ptr) { void default_multiplexer::exec_later(resumable* ptr) {
CAF_ASSERT(ptr); CAF_ASSERT(ptr);
CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0); CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0);
...@@ -764,20 +721,21 @@ void default_multiplexer::exec_later(resumable* ptr) { ...@@ -764,20 +721,21 @@ void default_multiplexer::exec_later(resumable* ptr) {
} }
connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
default_socket&& sock) { native_socket fd) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
class impl : public scribe { class impl : public scribe {
public: public:
impl(abstract_broker* ptr, default_socket&& s) impl(abstract_broker* ptr, default_multiplexer& ref, native_socket sockfd)
: scribe(ptr, network::conn_hdl_from_socket(s)), : scribe(ptr, network::conn_hdl_from_socket(sockfd)),
launched_(false), launched_(false),
stream_(s.backend()) { stream_(ref, sockfd) {
stream_.init(std::move(s)); // nop
} }
void configure_read(receive_policy::config config) override { void configure_read(receive_policy::config config) override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.configure_read(config); stream_.configure_read(config);
if (! launched_) launch(); if (! launched_)
launch();
} }
void ack_writes(bool enable) override { void ack_writes(bool enable) override {
CAF_LOG_TRACE(CAF_ARG(enable)); CAF_LOG_TRACE(CAF_ARG(enable));
...@@ -812,24 +770,23 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -812,24 +770,23 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
} }
private: private:
bool launched_; bool launched_;
stream<default_socket> stream_; stream stream_;
}; };
auto ptr = make_counted<impl>(self, std::move(sock)); auto ptr = make_counted<impl>(self, *this, fd);
self->add_scribe(ptr); self->add_scribe(ptr);
return ptr->hdl(); return ptr->hdl();
} }
accept_handle accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
default_multiplexer::add_tcp_doorman(abstract_broker* self, native_socket fd) {
default_socket_acceptor&& sock) {
CAF_LOG_TRACE(CAF_ARG(sock.fd())); CAF_LOG_TRACE(CAF_ARG(sock.fd()));
CAF_ASSERT(sock.fd() != network::invalid_native_socket); CAF_ASSERT(sock.fd() != network::invalid_native_socket);
class impl : public doorman { class impl : public doorman {
public: public:
impl(abstract_broker* ptr, default_socket_acceptor&& s) impl(abstract_broker* ptr, default_multiplexer& ref, native_socket sockfd)
: doorman(ptr, network::accept_hdl_from_socket(s)), : doorman(ptr, network::accept_hdl_from_socket(sockfd)),
acceptor_(s.backend()) { acceptor_(ref, sockfd) {
acceptor_.init(std::move(s)); // nop
} }
void new_connection() override { void new_connection() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -854,16 +811,16 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -854,16 +811,16 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
return local_port_of_fd(acceptor_.fd()); return local_port_of_fd(acceptor_.fd());
} }
private: private:
network::acceptor<default_socket_acceptor> acceptor_; network::acceptor acceptor_;
}; };
auto ptr = make_counted<impl>(self, std::move(sock)); auto ptr = make_counted<impl>(self, *this, fd);
self->add_doorman(ptr); self->add_doorman(ptr);
return ptr->hdl(); return ptr->hdl();
} }
connection_handle default_multiplexer::new_tcp_scribe(const std::string& host, connection_handle default_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) { uint16_t port) {
auto fd = new_tcp_connection_impl(host, port); auto fd = new_tcp_connection(host, port);
return connection_handle::from_int(int64_from_native_socket(fd)); return connection_handle::from_int(int64_from_native_socket(fd));
} }
...@@ -873,17 +830,12 @@ void default_multiplexer::assign_tcp_scribe(abstract_broker* self, ...@@ -873,17 +830,12 @@ void default_multiplexer::assign_tcp_scribe(abstract_broker* self,
add_tcp_scribe(self, static_cast<native_socket>(hdl.id())); add_tcp_scribe(self, static_cast<native_socket>(hdl.id()));
} }
connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(fd));
return add_tcp_scribe(self, default_socket{*this, fd});
}
connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
const std::string& host, const std::string& host,
uint16_t port) { uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(host) << CAF_ARG(port));
return add_tcp_scribe(self, new_tcp_connection(*this, host, port)); return add_tcp_scribe(self, new_tcp_connection(host, port));
} }
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
...@@ -899,17 +851,12 @@ void default_multiplexer::assign_tcp_doorman(abstract_broker* ptr, ...@@ -899,17 +851,12 @@ void default_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id())); add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id()));
} }
accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) {
return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
}
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* host, bool reuse_addr) { const char* host, bool reuse_addr) {
auto acceptor = new_tcp_acceptor(*this, port, host, reuse_addr); auto acceptor = new_tcp_acceptor_impl(port, host, reuse_addr);
auto bound_port = acceptor.second; auto bound_port = acceptor.second;
return {add_tcp_doorman(self, std::move(acceptor.first)), bound_port}; return {add_tcp_doorman(self, acceptor.first), bound_port};
} }
/****************************************************************************** /******************************************************************************
...@@ -975,52 +922,268 @@ bool try_accept(native_socket& result, native_socket fd) { ...@@ -975,52 +922,268 @@ bool try_accept(native_socket& result, native_socket fd) {
return true; return true;
} }
event_handler::event_handler(default_multiplexer& dm) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: backend_(dm), : eventbf_(0),
eventbf_(0) { fd_(sockfd),
// nop read_channel_closed_(false),
backend_(dm) {
CAF_LOG_TRACE(CAF_ARG(sockfd));
set_fd_flags();
} }
event_handler::~event_handler() { event_handler::~event_handler() {
// nop if (fd_ != invalid_native_socket)
closesocket(fd_);
} }
default_socket::default_socket(default_multiplexer& ref, native_socket sockfd) void event_handler::close_read_channel() {
: parent_(ref), if (fd_ == invalid_native_socket || read_channel_closed_)
fd_(sockfd) { return;
CAF_LOG_TRACE(CAF_ARG(sockfd)); ::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX
if (sockfd != invalid_native_socket) { read_channel_closed_ = true;
}
void event_handler::set_fd_flags() {
if (fd_ == invalid_native_socket)
return;
// enable nonblocking IO, disable Nagle's algorithm, and suppress SIGPIPE // enable nonblocking IO, disable Nagle's algorithm, and suppress SIGPIPE
nonblocking(fd_, true); nonblocking(fd_, true);
tcp_nodelay(fd_, true); tcp_nodelay(fd_, true);
allow_sigpipe(fd_, false); allow_sigpipe(fd_, false);
}
pipe_reader::pipe_reader(default_multiplexer& dm)
: event_handler(dm, invalid_native_socket) {
// nop
}
void pipe_reader::removed_from_loop(operation) {
// nop
}
resumable* pipe_reader::try_read_next() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto res = read(fd(), &ptrval, sizeof(ptrval));
# endif
if (res != sizeof(ptrval))
return nullptr;
return reinterpret_cast<resumable*>(ptrval);
}
void pipe_reader::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
auto cb = try_read_next();
switch (cb->resume(&backend(), backend().max_throughput())) {
case resumable::resume_later:
backend().exec_later(cb);
break;
case resumable::done:
intrusive_ptr_release(cb);
break;
default:
; // ignored
}
break;
}
default:
// nop (simply ignore errors)
break;
} }
} }
default_socket::default_socket(default_socket&& other) void pipe_reader::init(native_socket fd) {
: parent_(other.parent_), fd_ = fd;
fd_(other.fd_) {
other.fd_ = invalid_native_socket;
} }
default_socket& default_socket::operator=(default_socket&& other) { stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
std::swap(fd_, other.fd_); : event_handler(backend_ref, sockfd),
return *this; read_threshold_(1),
collected_(0),
ack_writes_(false),
writing_(false),
written_(0) {
configure_read(receive_policy::at_most(1024));
} }
default_socket::~default_socket() { void stream::start(const manager_ptr& mgr) {
if (fd_ != invalid_native_socket) { CAF_ASSERT(mgr != nullptr);
CAF_LOG_DEBUG("close socket:" << CAF_ARG(fd_)); reader_ = mgr;
closesocket(fd_); backend().add(operation::read, fd(), this);
read_loop();
}
void stream::configure_read(receive_policy::config config) {
rd_flag_ = config.first;
max_ = config.second;
}
void stream::ack_writes(bool x) {
ack_writes_ = x;
}
void stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void stream::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (! wr_offline_buf_.empty() && ! writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
write_loop();
} }
} }
void default_socket::close_read() { void stream::stop_reading() {
if (fd_ != invalid_native_socket) { CAF_LOG_TRACE("");
::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX close_read_channel();
backend().del(operation::read, fd(), this);
}
void stream::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
}
}
void stream::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
size_t rb; // read bytes
if (! read_some(rb, fd(),
rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
backend().del(operation::read, fd(), this);
} else if (rb > 0) {
collected_ += rb;
if (collected_ >= read_threshold_) {
reader_->consume(&backend(), rd_buf_.data(), collected_);
read_loop();
}
}
break;
}
case operation::write: {
size_t wb; // written bytes
if (! write_some(wb, fd(),
wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
write_loop();
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
}
void stream::read_loop() {
collected_ = 0;
switch (rd_flag_) {
case receive_policy_flag::exactly:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = max_;
break;
case receive_policy_flag::at_most:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (rd_buf_.size() != max_size)
rd_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
void stream::write_loop() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
written_ = 0;
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
}
}
acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
sock_(invalid_native_socket) {
// nop
}
void acceptor::start(const manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()));
CAF_ASSERT(mgr != nullptr);
mgr_ = mgr;
backend().add(operation::read, fd(), this);
}
void acceptor::stop_reading() {
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()));
close_read_channel();
backend().del(operation::read, fd(), this);
}
void acceptor::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_connection();
}
}
} }
} }
void acceptor::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(accept_sock_.fd()) << CAF_ARG(op));
if (op == operation::read)
mgr_.reset();
}
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(native_socket fd) : fd_(fd) { explicit socket_guard(native_socket fd) : fd_(fd) {
...@@ -1108,7 +1271,7 @@ bool ip_connect(native_socket fd, const std::string& host, uint16_t port) { ...@@ -1108,7 +1271,7 @@ bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
return connect(fd, reinterpret_cast<const sockaddr*>(&sa), sizeof(sa)) == 0; return connect(fd, reinterpret_cast<const sockaddr*>(&sa), sizeof(sa)) == 0;
} }
native_socket new_tcp_connection_impl(const std::string& host, uint16_t port, native_socket new_tcp_connection(const std::string& host, uint16_t port,
maybe<protocol> preferred) { maybe<protocol> preferred) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred)); CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred));
CAF_LOG_INFO("try to connect to:" << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_INFO("try to connect to:" << CAF_ARG(host) << CAF_ARG(port));
...@@ -1129,7 +1292,7 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port, ...@@ -1129,7 +1292,7 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
} }
sguard.close(); sguard.close();
// IPv4 fallback // IPv4 fallback
return new_tcp_connection_impl(host, port, ipv4); return new_tcp_connection(host, port, ipv4);
} }
if (! ip_connect<AF_INET>(fd, res->first, port)) { if (! ip_connect<AF_INET>(fd, res->first, port)) {
CAF_LOG_ERROR("could not connect to:" << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_ERROR("could not connect to:" << CAF_ARG(host) << CAF_ARG(port));
...@@ -1139,11 +1302,6 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port, ...@@ -1139,11 +1302,6 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
return sguard.release(); return sguard.release();
} }
default_socket new_tcp_connection(default_multiplexer& backend,
const std::string& host, uint16_t port) {
return default_socket{backend, new_tcp_connection_impl(host, port)};
}
template <class SockAddrType> template <class SockAddrType>
void read_port(native_socket fd, SockAddrType& sa) { void read_port(native_socket fd, SockAddrType& sa) {
socklen_t len = sizeof(SockAddrType); socklen_t len = sizeof(SockAddrType);
...@@ -1226,18 +1384,6 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) { ...@@ -1226,18 +1384,6 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
return {sguard.release(), p}; return {sguard.release(), p};
} }
std::pair<default_socket_acceptor, uint16_t>
new_tcp_acceptor(default_multiplexer& backend, uint16_t port,
const char* addr, bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")
<< ", " << CAF_ARG(reuse));
auto acceptor = new_tcp_acceptor_impl(port, addr, reuse);
auto bound_port = acceptor.second;
CAF_ASSERT(port == 0 || bound_port == port);
return {default_socket_acceptor{backend, std::move(acceptor.first)},
bound_port};
}
std::string local_addr_of_fd(native_socket fd) { std::string local_addr_of_fd(native_socket fd) {
sockaddr_storage st; sockaddr_storage st;
socklen_t st_len = sizeof(st); socklen_t st_len = sizeof(st);
...@@ -1250,7 +1396,8 @@ std::string local_addr_of_fd(native_socket fd) { ...@@ -1250,7 +1396,8 @@ std::string local_addr_of_fd(native_socket fd) {
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr, return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr)); addr, sizeof(addr));
case AF_INET6: case AF_INET6:
return inet_ntop(AF_INET6, &reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr, return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr)); addr, sizeof(addr));
default: default:
break; break;
...@@ -1278,7 +1425,8 @@ std::string remote_addr_of_fd(native_socket fd) { ...@@ -1278,7 +1425,8 @@ std::string remote_addr_of_fd(native_socket fd) {
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr, return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr)); addr, sizeof(addr));
case AF_INET6: case AF_INET6:
return inet_ntop(AF_INET6, &reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr, return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr)); addr, sizeof(addr));
default: default:
break; break;
......
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