From 3f55065e433e600fbab5aadb21a819b079839067 Mon Sep 17 00:00:00 2001 From: halx99 Date: Wed, 28 Dec 2022 02:43:26 +0800 Subject: [PATCH 01/19] Implement reactor by poll Notes: - Windows 8.1+ required - Unix all --- yasio/yasio.cpp | 158 +++++++++++++++++++++++++++++------------------- yasio/yasio.hpp | 21 ++++--- 2 files changed, 108 insertions(+), 71 deletions(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 1024b3b16..29d3148a5 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -189,6 +189,23 @@ static yasio__global_state& yasio__shared_globals(const print_fn2_t& prt = nullp static yasio__global_state __global_state(prt); return __global_state; } +static void pollfd_mod(socket_native_type fd, std::vector& fds, int add_flags, int remove_flags) +{ + auto it = std::find_if(fds.begin(), fds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); + if (it != fds.end()) + { + it->events |= add_flags; + it->events &= ~remove_flags; + if (it->events == 0) + fds.erase(it); + } + else + { + auto combined_flags = add_flags & ~remove_flags; + if (combined_flags) + fds.push_back(pollfd{fd, static_cast(combined_flags), 0}); + } +} } // namespace /// highp_timer @@ -815,11 +832,6 @@ void io_service::initialize(const io_hostent* channel_eps, int channel_count) if (channel_count < 1) channel_count = 1; - FD_ZERO(&fds_array_[read_op]); - FD_ZERO(&fds_array_[write_op]); - FD_ZERO(&fds_array_[except_op]); - - this->max_nfds_ = 0; options_.resolv_ = [=](std::vector& eps, const char* host, unsigned short port) { return this->resolve(eps, host, port); }; register_descriptor(interrupter_.read_descriptor(), YEM_POLLIN); @@ -898,13 +910,14 @@ void io_service::run() #endif #if defined(YASIO_HAVE_CARES) recreate_ares_channel(); + ares_socket_t ares_socks[ARES_GETSOCK_MAXNUM] = {0}; + int ares_socks_count = 0; #endif // Call once at startup this->ipsv_ = static_cast(xxsocket::getipsv()); // The core event loop - fd_set fds_array[max_ops]; this->wait_duration_ = YASIO_MAX_WAIT_DURATION; do { @@ -912,7 +925,30 @@ void io_service::run() this->wait_duration_ = YASIO_MAX_WAIT_DURATION; // Reset next wait duration if (wait_duration > 0) { - int retval = do_select(fds_array, wait_duration); + this->pollfds = this->pollfd_registry_; + timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; +#if defined(YASIO_HAVE_CARES) + if (this->ares_outstanding_work_ > 0) + { + int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); + ares_socks_count = 0; + for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) + { + if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) + { + auto fd = ares_socks[i]; + ++ares_socks_count; + pollfd_mod(fd, POLLIN | POLLOUT, 0, false); + } + else + break; + } + ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); + } +#endif + YASIO_KLOGV("[core] poll waiting... %ld milliseconds", waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); + int retval = ::poll(this->pollfds.data(), this->pollfds.size(), static_cast(wait_duration / 1000)); + YASIO_KLOGV("[core] poll waked up, retval=%d", retval); if (retval < 0) { int ec = xxsocket::get_last_errno(); @@ -924,7 +960,7 @@ void io_service::run() if (retval == 0) YASIO_KLOGV("[core] %s", "do_select is timeout, process_timers()"); - else if (FD_ISSET(this->interrupter_.read_descriptor(), &(fds_array[read_op]))) + else if (pollfd_isset(this->interrupter_.read_descriptor(), POLLIN)) { // Reset the interrupter. if (!interrupter_.reset()) interrupter_.recreate(); @@ -934,14 +970,14 @@ void io_service::run() #if defined(YASIO_HAVE_CARES) // process possible async resolve requests. - process_ares_requests(fds_array); + process_ares_requests(ares_socks, ares_socks_count); #endif // process active transports - process_transports(fds_array); + process_transports(); // process active channels - process_channels(fds_array); + process_channels(); // process timeout timers process_timers(); @@ -956,13 +992,13 @@ void io_service::run() this->state_ = io_service::state::AT_EXITING; } -void io_service::process_transports(fd_set* fds_array) +void io_service::process_transports() { // preform transports for (auto iter = transports_.begin(); iter != transports_.end();) { auto transport = *iter; - bool ok = (do_read(transport, fds_array) && do_write(transport)); + bool ok = (do_read(transport) && do_write(transport)); if (ok) { int opm = transport->opmask_ | transport->ctx_->opmask_ | this->stop_flag_; @@ -978,7 +1014,7 @@ void io_service::process_transports(fd_set* fds_array) iter = transports_.erase(iter); } } -void io_service::process_channels(fd_set* fds_array) +void io_service::process_channels() { if (!this->channel_ops_.empty()) { @@ -1004,7 +1040,7 @@ void io_service::process_channels(fd_set* fds_array) handle_connect_failed(ctx, ctx->error_); } else if (ctx->state_ == io_base::state::CONNECTING) - do_connect_completion(ctx, fds_array); + do_connect_completion(ctx); finish = ctx->error_ != EINPROGRESS; } else if (yasio__testbits(ctx->properties_, YCM_SERVER)) @@ -1017,7 +1053,7 @@ void io_service::process_channels(fd_set* fds_array) finish = (ctx->state_ != io_base::state::OPENED); if (!finish) - do_accept_completion(ctx, fds_array); + do_accept_completion(ctx); else ctx->bytes_transferred_ = 0; } @@ -1094,28 +1130,54 @@ void io_service::handle_close(transport_handle_t thandle) } void io_service::register_descriptor(const socket_native_type fd, int flags) { + int underlying_flags = 0; if (yasio__testbits(flags, YEM_POLLIN)) - FD_SET(fd, &(fds_array_[read_op])); + underlying_flags |= POLLIN; if (yasio__testbits(flags, YEM_POLLOUT)) - FD_SET(fd, &(fds_array_[write_op])); + underlying_flags |= POLLOUT; if (yasio__testbits(flags, YEM_POLLERR)) - FD_SET(fd, &(fds_array_[except_op])); - - if (max_nfds_ < static_cast(fd) + 1) - max_nfds_ = static_cast(fd) + 1; + underlying_flags |= POLLERR; + pollfd_mod(fd, underlying_flags, 0); } void io_service::unregister_descriptor(const socket_native_type fd, int flags) { + int underlying_flags = 0; if (yasio__testbits(flags, YEM_POLLIN)) - FD_CLR(fd, &(fds_array_[read_op])); + underlying_flags |= POLLIN; if (yasio__testbits(flags, YEM_POLLOUT)) - FD_CLR(fd, &(fds_array_[write_op])); + underlying_flags |= POLLOUT; if (yasio__testbits(flags, YEM_POLLERR)) - FD_CLR(fd, &(fds_array_[except_op])); + underlying_flags |= POLLERR; + + pollfd_mod(fd, 0, underlying_flags); +} +void io_service::pollfd_mod(socket_native_type fd, int add_flags, int remove_flags, bool use_registry) +{ + auto& fds = use_registry ? this->pollfd_registry_ : this->pollfds; + auto it = std::find_if(fds.begin(), fds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); + if (it != fds.end()) + { + // POLLIN + it->events |= add_flags; + it->events &= ~remove_flags; + if (it->events == 0) + fds.erase(it); + } + else + { + auto combined_flags = add_flags & ~remove_flags; + if (combined_flags) + fds.push_back(pollfd{fd, static_cast(combined_flags), 0}); + } +} +int io_service::pollfd_isset(socket_native_type fd, int flags) const +{ + auto it = std::find_if(this->pollfds.begin(), this->pollfds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); + return it != this->pollfds.end() ? (it->revents & flags) : 0; } int io_service::write(transport_handle_t transport, dynamic_buffer_t buffer, completion_cb_t handler) { @@ -1214,7 +1276,7 @@ void io_service::do_connect(io_channel* ctx) this->handle_connect_failed(ctx, xxsocket::get_last_errno()); } -void io_service::do_connect_completion(io_channel* ctx, fd_set* fds_array) +void io_service::do_connect_completion(io_channel* ctx) { assert(ctx->state_ == io_base::state::CONNECTING); if (ctx->state_ == io_base::state::CONNECTING) @@ -1236,7 +1298,7 @@ void io_service::do_connect_completion(io_channel* ctx, fd_set* fds_array) #else if (!yasio__testbits(ctx->properties_, YCPF_SSL_HANDSHAKING)) { - if (FD_ISSET(ctx->socket_->native_handle(), &fds_array[write_op]) || FD_ISSET(ctx->socket_->native_handle(), &fds_array[read_op])) + if (pollfd_isset(ctx->socket_->native_handle(), POLLIN | POLLOUT)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { @@ -1455,22 +1517,14 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -void io_service::process_ares_requests(fd_set* fds_array) +void io_service::process_ares_requests(socket_native_type* socks, int count) { if (this->ares_outstanding_work_ > 0) { - ares_socket_t socks[ARES_GETSOCK_MAXNUM] = {0}; - int bitmask = ::ares_getsock(this->ares_, socks, ARES_GETSOCK_MAXNUM); - - for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) + for (auto i = 0; i < count; ++i) { - if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) - { - auto fd = socks[i]; - ::ares_process_fd(this->ares_, FD_ISSET(fd, &(fds_array[read_op])) ? fd : ARES_SOCKET_BAD, FD_ISSET(fd, &(fds_array[write_op])) ? fd : ARES_SOCKET_BAD); - } - else - break; + auto fd = socks[i]; + ::ares_process_fd(this->ares_, pollfd_isset(fd, POLLIN) ? fd : ARES_SOCKET_BAD, pollfd_isset(fd, POLLOUT) ? fd : ARES_SOCKET_BAD); } } } @@ -1600,12 +1654,12 @@ void io_service::do_accept(io_channel* ctx) handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx, 1)); #endif } -void io_service::do_accept_completion(io_channel* ctx, fd_set* fds_array) +void io_service::do_accept_completion(io_channel* ctx) { if (ctx->state_ == io_base::state::OPENED) { int error = 0; - if (FD_ISSET(ctx->socket_->native_handle(), &fds_array[read_op]) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) + if (pollfd_isset(ctx->socket_->native_handle(), POLLIN) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { if (yasio__testbits(ctx->properties_, YCM_TCP)) { @@ -1791,7 +1845,7 @@ void io_service::handle_connect_failed(io_channel* ctx, int error) YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx->index_, ctx->format_destination().c_str(), error, io_service::strerror(error)); handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx)); } -bool io_service::do_read(transport_handle_t transport, fd_set* fds_array) +bool io_service::do_read(transport_handle_t transport) { bool ret = false; do @@ -1799,7 +1853,7 @@ bool io_service::do_read(transport_handle_t transport, fd_set* fds_array) if (!transport->socket_->is_open()) break; int error = 0; - int revent = FD_ISSET(transport->socket_->native_handle(), &(fds_array[read_op])); + int revent = pollfd_isset(transport->socket_->native_handle(), POLLIN | POLLHUP | POLLERR | POLLNVAL); int n = transport->do_read(revent, error, this->wait_duration_); if (n >= 0) { @@ -1969,24 +2023,6 @@ void io_service::process_timers() if (n) sort_timers(); } -int io_service::do_select(fd_set* fdsa, highp_time_t wait_duration) -{ - ::memcpy(fdsa, this->fds_array_, sizeof(this->fds_array_)); - timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; -#if defined(YASIO_HAVE_CARES) - int nfds = -1; - if (this->ares_outstanding_work_ > 0 && (nfds = ::ares_fds(this->ares_, &fdsa[read_op], &fdsa[write_op])) > 0) - { - ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); - if (this->max_nfds_ < nfds) - this->max_nfds_ = nfds; - } -#endif - YASIO_KLOGV("[core] socket.select max_nfds_:%d waiting... %ld milliseconds", max_nfds_, waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); - int retval = ::select(this->max_nfds_, &(fdsa[read_op]), &(fdsa[write_op]), nullptr, &waitd_tv); - YASIO_KLOGV("[core] socket.select waked up, retval=%d", retval); - return retval; -} highp_time_t io_service::get_timeout(highp_time_t usec) { if (this->timer_queue_.empty()) diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 86a90ffea..5b289f534 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1062,19 +1062,19 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL bool open_internal(io_channel*); - YASIO__DECL void process_transports(fd_set* fds_array); - YASIO__DECL void process_channels(fd_set* fds_array); + YASIO__DECL void process_transports(); + YASIO__DECL void process_channels(); YASIO__DECL void process_timers(); YASIO__DECL void interrupt(); YASIO__DECL highp_time_t get_timeout(highp_time_t usec); - YASIO__DECL int do_select(fd_set* fds_array, highp_time_t wait_duration); + YASIO__DECL int poll_wait(highp_time_t wait_duration, socket_native_type* ares_socks); YASIO__DECL int do_resolve(io_channel* ctx); YASIO__DECL void do_connect(io_channel*); - YASIO__DECL void do_connect_completion(io_channel*, fd_set* fds_array); + YASIO__DECL void do_connect_completion(io_channel*); #if defined(YASIO_SSL_BACKEND) YASIO__DECL void init_ssl_context(); @@ -1086,7 +1086,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL void process_ares_requests(fd_set* fds_array); + YASIO__DECL void process_ares_requests(socket_native_type* socks, int count); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); YASIO__DECL void destroy_ares_channel(); @@ -1102,11 +1102,13 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL void register_descriptor(const socket_native_type fd, int flags); YASIO__DECL void unregister_descriptor(const socket_native_type fd, int flags); + YASIO__DECL void pollfd_mod(socket_native_type fd, int add_flags, int remove_flags, bool use_registry = true); + YASIO__DECL int pollfd_isset(socket_native_type fd, int flags) const; // The major non-blocking event-loop YASIO__DECL void run(void); - YASIO__DECL bool do_read(transport_handle_t, fd_set* fds_array); + YASIO__DECL bool do_read(transport_handle_t); bool do_write(transport_handle_t transport) { return transport->do_write(this->wait_duration_); } YASIO__DECL void unpack(transport_handle_t, int bytes_expected, int bytes_transferred, int bytes_to_strip); @@ -1129,7 +1131,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // supporting server YASIO__DECL void do_accept(io_channel*); - YASIO__DECL void do_accept_completion(io_channel*, fd_set* fds_array); + YASIO__DECL void do_accept_completion(io_channel*); YASIO__DECL static const char* strerror(int error); @@ -1172,8 +1174,6 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // the next wait duration for socket.select highp_time_t wait_duration_; - // the max nfds for socket.select, must be max_fd + 1 - int max_nfds_; enum { read_op, @@ -1181,7 +1181,8 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] except_op, max_ops, }; - fd_set fds_array_[max_ops]; + std::vector pollfd_registry_; + std::vector pollfds; // options struct __unnamed_options { From 0441dac39704c425e0c1afb01f21b6f63ae2228b Mon Sep 17 00:00:00 2001 From: halx99 Date: Wed, 28 Dec 2022 02:48:04 +0800 Subject: [PATCH 02/19] Fix ci --- yasio/yasio.cpp | 2 +- yasio/yasio.hpp | 2 -- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 29d3148a5..30d134c7f 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -1283,7 +1283,7 @@ void io_service::do_connect_completion(io_channel* ctx) { int error = -1; #if !defined(YASIO_SSL_BACKEND) - if (FD_ISSET(ctx->socket_->native_handle(), &fds_array[write_op]) || FD_ISSET(ctx->socket_->native_handle(), &fds_array[read_op])) + if (pollfd_isset(ctx->socket_->native_handle(), POLLIN | POLLOUT)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 5b289f534..e5a829a71 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1070,8 +1070,6 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL highp_time_t get_timeout(highp_time_t usec); - YASIO__DECL int poll_wait(highp_time_t wait_duration, socket_native_type* ares_socks); - YASIO__DECL int do_resolve(io_channel* ctx); YASIO__DECL void do_connect(io_channel*); YASIO__DECL void do_connect_completion(io_channel*); From d482ff46bcae5e9803232c2cab58c76f7809ddd2 Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 21:05:52 +0800 Subject: [PATCH 03/19] Improve file descriptor set management --- yasio/detail/config.hpp | 12 ++- yasio/detail/poll_fd_set.hpp | 98 ++++++++++++++++++++++ yasio/detail/select_fd_set.hpp | 86 +++++++++++++++++++ yasio/detail/socket.hpp | 16 ++++ yasio/yasio.cpp | 148 +++++++++------------------------ yasio/yasio.hpp | 25 ++---- 6 files changed, 257 insertions(+), 128 deletions(-) create mode 100644 yasio/detail/poll_fd_set.hpp create mode 100644 yasio/detail/select_fd_set.hpp diff --git a/yasio/detail/config.hpp b/yasio/detail/config.hpp index ce2da5563..281ff64fc 100644 --- a/yasio/detail/config.hpp +++ b/yasio/detail/config.hpp @@ -192,7 +192,7 @@ SOFTWARE. /* ** The yasio version macros */ -#define YASIO_VERSION_NUM 0x033905 +#define YASIO_VERSION_NUM 0x033906 /* ** The macros used by io_service. @@ -232,4 +232,14 @@ SOFTWARE. #define YASIO_SSL_PIN "yasio_ssl_client" #define YASIO_SSL_PIN_LEN (sizeof(YASIO_SSL_PIN) - 1) + +/* +** yasio bitop macros +*/ + +#define yasio__setbits(x, m) ((x) |= (m)) +#define yasio__clearbits(x, m) ((x) &= ~(m)) +#define yasio__testbits(x, m) ((x) & (m)) +#define yasio__setlobyte(x, v) ((x) = ((x) & ~((decltype(x))0xff)) | (v)) + #endif diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp new file mode 100644 index 000000000..f33b110b7 --- /dev/null +++ b/yasio/detail/poll_fd_set.hpp @@ -0,0 +1,98 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// A multi-platform support c++11 library with focus on asynchronous socket I/O for any +// client application. +////////////////////////////////////////////////////////////////////////////////////////// +// +// detail/poll_event_registry.hpp +// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) + +#pragma once +#include +#include "yasio/detail/socket.hpp" + +namespace yasio +{ +YASIO__NS_INLINE +namespace inet +{ +class poll_fd_set { +public: + poll_fd_set& operator=(poll_fd_set& rhs) + { + this->fd_set_ = rhs.fd_set_; + return *this; + } + + int poll_wait(long long timeout_usec) { return ::poll(this->fd_set_.data(), static_cast(this->fd_set_.size()), timeout_usec / 1000); } + + int has_events(socket_native_type fd, int events) const + { + int underlying_events = 0; + if (events & socket_event::read) + underlying_events |= POLLIN; + if (events & socket_event::write) + underlying_events |= POLLOUT; + if (events & socket_event::error) + underlying_events |= (POLLERR | POLLHUP | POLLNVAL); + auto it = std::find_if(this->fd_set_.begin(), this->fd_set_.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); + return it != this->fd_set_.end() ? (it->revents & underlying_events) : 0; + } + + void reset() { this->fd_set_.clear(); } + + void register_descriptor(socket_native_type fd, int events) + { + int underlying_flags = 0; + if (yasio__testbits(events, socket_event::read)) + underlying_flags |= POLLIN; + + if (yasio__testbits(events, socket_event::write)) + underlying_flags |= POLLOUT; + + if (yasio__testbits(events, socket_event::error)) + underlying_flags |= POLLERR; + pollfd_mod(this->fd_set_, fd, underlying_flags, 0); + } + + void deregister_descriptor(socket_native_type fd, int events) + { + int underlying_flags = 0; + if (yasio__testbits(events, socket_event::read)) + underlying_flags |= POLLIN; + + if (yasio__testbits(events, socket_event::write)) + underlying_flags |= POLLOUT; + + if (yasio__testbits(events, socket_event::error)) + underlying_flags |= POLLERR; + + pollfd_mod(this->fd_set_, fd, 0, underlying_flags); + } + +protected: + static void pollfd_mod(std::vector& fdset, socket_native_type fd, int add_flags, int remove_flags) + { + auto it = std::find_if(fdset.begin(), fdset.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); + if (it != fdset.end()) + { + // POLLIN + it->events |= add_flags; + it->events &= ~remove_flags; + if (it->events == 0) + fdset.erase(it); + } + else + { + auto combined_flags = add_flags & ~remove_flags; + if (combined_flags) + fdset.push_back(pollfd{fd, static_cast(combined_flags), 0}); + } + } + +protected: + std::vector fd_set_; +}; +} // namespace inet +} // namespace yasio diff --git a/yasio/detail/select_fd_set.hpp b/yasio/detail/select_fd_set.hpp new file mode 100644 index 000000000..90c62d87b --- /dev/null +++ b/yasio/detail/select_fd_set.hpp @@ -0,0 +1,86 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// A multi-platform support c++11 library with focus on asynchronous socket I/O for any +// client application. +////////////////////////////////////////////////////////////////////////////////////////// +// +// detail/poll_reactor.hpp +// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) +#pragma once + +#include +#include "yasio/detail/socket.hpp" + +namespace yasio +{ +YASIO__NS_INLINE +namespace inet +{ +class select_fd_set { +public: + select_fd_set() { + reset(); + } + // TODO: cb + int poll_events(int timeout, select_fd_set& revents) + { + // std::copy(this->fd_set_.begin(), this->fd_set_.end(), std::back_inserter(revents.fd_set_)); + // return ::poll(revents, revents.count(), timeout); + } + + int has_events(socket_native_type fd, int events) const + { + + } + + void reset() + { + FD_ZERO(&fd_set_[read_op]); + FD_ZERO(&fd_set_[write_op]); + FD_ZERO(&fd_set_[except_op]); + max_nfds_ = 0; + } + + void register_descriptor(socket_native_type fd, int events) + { + if (yasio__testbits(events, YEM_POLLIN)) + FD_SET(fd, &(fd_set_[read_op])); + + if (yasio__testbits(events, YEM_POLLOUT)) + FD_SET(fd, &(fd_set_[write_op])); + + if (yasio__testbits(events, YEM_POLLERR)) + FD_SET(fd, &(fd_set_[except_op])); + + if (max_nfds_ < static_cast(fd) + 1) + max_nfds_ = static_cast(fd) + 1; + } + + void deregister_descriptor(socket_native_type fd, int events) + { + if (yasio__testbits(events, YEM_POLLIN)) + FD_CLR(fd, &(fd_set_[read_op])); + + if (yasio__testbits(events, YEM_POLLOUT)) + FD_CLR(fd, &(fd_set_[write_op])); + + if (yasio__testbits(events, YEM_POLLERR)) + FD_CLR(fd, &(fd_set_[except_op])); + } + +protected: + +protected: + enum + { + read_op, + write_op, + except_op, + max_ops, + }; + fd_set fd_set_[max_ops]; + int max_nfds_ = 0; +}; +} // namespace inet +} // namespace yasio diff --git a/yasio/detail/socket.hpp b/yasio/detail/socket.hpp index 0f4467804..d682b3974 100644 --- a/yasio/detail/socket.hpp +++ b/yasio/detail/socket.hpp @@ -228,4 +228,20 @@ inline bool IN6_IS_ADDR_GLOBAL(const in6_addr* a) #define YASIO_ADDR_ANY(af) (af == AF_INET ? "0.0.0.0" : "::") +namespace yasio +{ +YASIO__NS_INLINE +namespace inet +{ +struct socket_event { + enum + { // event mask + read = 1, + write = 2, + error = 4, + readwrite = read | write, + }; +}; +} +} // namespace yasio #endif diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 729e38583..fddd0e51e 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -76,11 +76,6 @@ SOFTWARE. # define YASIO_KLOGV(format, ...) YASIO_KLOG_CP(YLOG_V, format, ##__VA_ARGS__) #endif -#define yasio__setbits(x, m) ((x) |= (m)) -#define yasio__clearbits(x, m) ((x) &= ~(m)) -#define yasio__testbits(x, m) ((x) & (m)) -#define yasio__setlobyte(x, v) ((x) = ((x) & ~((decltype(x))0xff)) | (v)) - #if defined(_MSC_VER) # pragma warning(push) # pragma warning(disable : 6320 6322 4996) @@ -94,12 +89,6 @@ namespace inet { namespace { -enum -{ // event mask - YEM_POLLIN = 1, - YEM_POLLOUT = 2, - YEM_POLLERR = 4, -}; enum : uint8_t { // op masks and stop flags @@ -189,23 +178,6 @@ static yasio__global_state& yasio__shared_globals(const print_fn2_t& prt = nullp static yasio__global_state __global_state(prt); return __global_state; } -static void pollfd_mod(socket_native_type fd, std::vector& fds, int add_flags, int remove_flags) -{ - auto it = std::find_if(fds.begin(), fds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); - if (it != fds.end()) - { - it->events |= add_flags; - it->events &= ~remove_flags; - if (it->events == 0) - fds.erase(it); - } - else - { - auto combined_flags = add_flags & ~remove_flags; - if (combined_flags) - fds.push_back(pollfd{fd, static_cast(combined_flags), 0}); - } -} } // namespace /// highp_timer @@ -403,7 +375,7 @@ bool io_transport::do_write(highp_time_t& wait_duration) { // system kernel buffer full if (!pollout_registerred_) { - get_service().register_descriptor(socket_->native_handle(), YEM_POLLOUT); + get_service().register_descriptor(socket_->native_handle(), socket_event::write); pollout_registerred_ = true; } } @@ -412,7 +384,7 @@ bool io_transport::do_write(highp_time_t& wait_duration) } if (no_wevent && pollout_registerred_) { - get_service().unregister_descriptor(socket_->native_handle(), YEM_POLLOUT); + get_service().unregister_descriptor(socket_->native_handle(), socket_event::write); pollout_registerred_ = false; } ret = true; @@ -833,7 +805,7 @@ void io_service::initialize(const io_hostent* channel_eps, int channel_count) channel_count = 1; options_.resolv_ = [=](std::vector& eps, const char* host, unsigned short port) { return this->resolve(eps, host, port); }; - register_descriptor(interrupter_.read_descriptor(), YEM_POLLIN); + register_descriptor(interrupter_.read_descriptor(), socket_event::read); // create channels create_channels(channel_eps, channel_count); @@ -853,7 +825,7 @@ void io_service::finalize() life_token_.reset(); #endif destroy_channels(); - unregister_descriptor(interrupter_.read_descriptor(), YEM_POLLIN); + unregister_descriptor(interrupter_.read_descriptor(), socket_event::read); options_.on_event_ = nullptr; options_.resolv_ = nullptr; @@ -919,13 +891,16 @@ void io_service::run() // The core event loop this->wait_duration_ = YASIO_MAX_WAIT_DURATION; + + poll_fd_set revents; // file_descriptor_set + do { auto wait_duration = get_timeout(this->wait_duration_); // Gets current wait duration this->wait_duration_ = YASIO_MAX_WAIT_DURATION; // Reset next wait duration if (wait_duration > 0) { - this->pollfds = this->pollfd_registry_; + revents = this->fd_set_; timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; #if defined(YASIO_HAVE_CARES) if (this->ares_outstanding_work_ > 0) @@ -938,7 +913,7 @@ void io_service::run() { auto fd = ares_socks[i]; ++ares_socks_count; - pollfd_mod(fd, POLLIN | POLLOUT, 0, false); + revents.register_descriptor(fd, socket_event::readwrite); } else break; @@ -946,9 +921,10 @@ void io_service::run() ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); } #endif - YASIO_KLOGV("[core] poll waiting... %ld milliseconds", waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); - int retval = ::poll(this->pollfds.data(), this->pollfds.size(), static_cast(wait_duration / 1000)); - YASIO_KLOGV("[core] poll waked up, retval=%d", retval); + int retval = revents.poll_wait(static_cast(wait_duration / 1000)); + //YASIO_KLOGV("[core] poll waiting... %ld milliseconds", waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); + //int retval = ::poll(this->pollfds.data(), this->pollfds.size(), static_cast(wait_duration / 1000)); + //YASIO_KLOGV("[core] poll waked up, retval=%d", retval); if (retval < 0) { int ec = xxsocket::get_last_errno(); @@ -960,7 +936,7 @@ void io_service::run() if (retval == 0) YASIO_KLOGV("[core] %s", "do_select is timeout, process_timers()"); - else if (pollfd_isset(this->interrupter_.read_descriptor(), POLLIN)) + else if (revents.has_events(this->interrupter_.read_descriptor(), socket_event::read)) { // Reset the interrupter. if (!interrupter_.reset()) interrupter_.recreate(); @@ -970,14 +946,14 @@ void io_service::run() #if defined(YASIO_HAVE_CARES) // process possible async resolve requests. - process_ares_requests(ares_socks, ares_socks_count); + process_ares_requests(ares_socks, ares_socks_count, revents); #endif // process active transports - process_transports(); + process_transports(revents); // process active channels - process_channels(); + process_channels(revents); // process timeout timers process_timers(); @@ -992,13 +968,13 @@ void io_service::run() this->state_ = io_service::state::AT_EXITING; } -void io_service::process_transports() +void io_service::process_transports(poll_fd_set& revents) { // preform transports for (auto iter = transports_.begin(); iter != transports_.end();) { auto transport = *iter; - bool ok = (do_read(transport) && do_write(transport)); + bool ok = (do_read(transport, revents) && do_write(transport)); if (ok) { int opm = transport->opmask_ | transport->ctx_->opmask_ | this->stop_flag_; @@ -1014,7 +990,7 @@ void io_service::process_transports() iter = transports_.erase(iter); } } -void io_service::process_channels() +void io_service::process_channels(poll_fd_set& revents) { if (!this->channel_ops_.empty()) { @@ -1040,7 +1016,7 @@ void io_service::process_channels() handle_connect_failed(ctx, ctx->error_); } else if (ctx->state_ == io_base::state::CONNECTING) - do_connect_completion(ctx); + do_connect_completion(ctx, revents); finish = ctx->error_ != EINPROGRESS; } else if (yasio__testbits(ctx->properties_, YCM_SERVER)) @@ -1053,7 +1029,7 @@ void io_service::process_channels() finish = (ctx->state_ != io_base::state::OPENED); if (!finish) - do_accept_completion(ctx); + do_accept_completion(ctx, revents); else ctx->bytes_transferred_ = 0; } @@ -1128,57 +1104,9 @@ void io_service::handle_close(transport_handle_t thandle) cleanup_channel(ctx, false); } } -void io_service::register_descriptor(const socket_native_type fd, int flags) -{ - int underlying_flags = 0; - if (yasio__testbits(flags, YEM_POLLIN)) - underlying_flags |= POLLIN; - - if (yasio__testbits(flags, YEM_POLLOUT)) - underlying_flags |= POLLOUT; +void io_service::register_descriptor(const socket_native_type fd, int flags) { this->fd_set_.register_descriptor(fd, flags); } +void io_service::unregister_descriptor(const socket_native_type fd, int flags) { this->fd_set_.deregister_descriptor(fd, flags); } - if (yasio__testbits(flags, YEM_POLLERR)) - underlying_flags |= POLLERR; - pollfd_mod(fd, underlying_flags, 0); -} -void io_service::unregister_descriptor(const socket_native_type fd, int flags) -{ - int underlying_flags = 0; - if (yasio__testbits(flags, YEM_POLLIN)) - underlying_flags |= POLLIN; - - if (yasio__testbits(flags, YEM_POLLOUT)) - underlying_flags |= POLLOUT; - - if (yasio__testbits(flags, YEM_POLLERR)) - underlying_flags |= POLLERR; - - pollfd_mod(fd, 0, underlying_flags); -} -void io_service::pollfd_mod(socket_native_type fd, int add_flags, int remove_flags, bool use_registry) -{ - auto& fds = use_registry ? this->pollfd_registry_ : this->pollfds; - auto it = std::find_if(fds.begin(), fds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); - if (it != fds.end()) - { - // POLLIN - it->events |= add_flags; - it->events &= ~remove_flags; - if (it->events == 0) - fds.erase(it); - } - else - { - auto combined_flags = add_flags & ~remove_flags; - if (combined_flags) - fds.push_back(pollfd{fd, static_cast(combined_flags), 0}); - } -} -int io_service::pollfd_isset(socket_native_type fd, int flags) const -{ - auto it = std::find_if(this->pollfds.begin(), this->pollfds.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); - return it != this->pollfds.end() ? (it->revents & flags) : 0; -} int io_service::write(transport_handle_t transport, dynamic_buffer_t buffer, completion_cb_t handler) { if (transport && transport->is_open()) @@ -1258,7 +1186,7 @@ void io_service::do_connect(io_channel* ctx) else { ctx->set_last_errno(EINPROGRESS); - register_descriptor(ctx->socket_->native_handle(), YEM_POLLIN | YEM_POLLOUT); + register_descriptor(ctx->socket_->native_handle(), socket_event::readwrite); ctx->timer_.expires_from_now(std::chrono::microseconds(options_.connect_timeout_)); ctx->timer_.async_wait_once(*this, [ctx](io_service& thiz) { if (ctx->state_ != io_base::state::OPENED) @@ -1268,7 +1196,7 @@ void io_service::do_connect(io_channel* ctx) } else if (ret == 0) { // connect server successful immediately. - register_descriptor(ctx->socket_->native_handle(), YEM_POLLIN); + register_descriptor(ctx->socket_->native_handle(), socket_event::read); handle_connect_succeed(ctx, ctx->socket_); } // !!!NEVER GO HERE } @@ -1276,7 +1204,7 @@ void io_service::do_connect(io_channel* ctx) this->handle_connect_failed(ctx, xxsocket::get_last_errno()); } -void io_service::do_connect_completion(io_channel* ctx) +void io_service::do_connect_completion(io_channel* ctx, poll_fd_set& fd_set) { assert(ctx->state_ == io_base::state::CONNECTING); if (ctx->state_ == io_base::state::CONNECTING) @@ -1298,12 +1226,12 @@ void io_service::do_connect_completion(io_channel* ctx) #else if (!yasio__testbits(ctx->properties_, YCPF_SSL_HANDSHAKING)) { - if (pollfd_isset(ctx->socket_->native_handle(), POLLIN | POLLOUT)) + if (fd_set.has_events(ctx->socket_->native_handle(), socket_event::readwrite)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { // The nonblocking tcp handshake complete, remove write event avoid high-CPU occupation - unregister_descriptor(ctx->socket_->native_handle(), YEM_POLLOUT); + unregister_descriptor(ctx->socket_->native_handle(), socket_event::write); if (!yasio__testbits(ctx->properties_, YCM_SSL)) handle_connect_succeed(ctx, ctx->socket_); else @@ -1517,14 +1445,14 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -void io_service::process_ares_requests(socket_native_type* socks, int count) +void io_service::process_ares_requests(socket_native_type* socks, int count, poll_fd_set& revents) { if (this->ares_outstanding_work_ > 0) { for (auto i = 0; i < count; ++i) { auto fd = socks[i]; - ::ares_process_fd(this->ares_, pollfd_isset(fd, POLLIN) ? fd : ARES_SOCKET_BAD, pollfd_isset(fd, POLLOUT) ? fd : ARES_SOCKET_BAD); + ::ares_process_fd(this->ares_, revents.has_events(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, revents.has_events(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); } } } @@ -1638,7 +1566,7 @@ void io_service::do_accept(io_channel* ctx) ctx->join_multicast_group(); ctx->buffer_.resize(YASIO_INET_BUFFER_SIZE); } - register_descriptor(ctx->socket_->native_handle(), YEM_POLLIN); + register_descriptor(ctx->socket_->native_handle(), socket_event::read); YASIO_KLOGD("[index: %d] open server succeed, socket.fd=%d listening at %s...", ctx->index_, (int)ctx->socket_->native_handle(), ep.to_string().c_str()); error = 0; } while (false); @@ -1654,12 +1582,12 @@ void io_service::do_accept(io_channel* ctx) handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx, 1)); #endif } -void io_service::do_accept_completion(io_channel* ctx) +void io_service::do_accept_completion(io_channel* ctx, poll_fd_set& revents) { if (ctx->state_ == io_base::state::OPENED) { int error = 0; - if (pollfd_isset(ctx->socket_->native_handle(), POLLIN) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) + if (revents.has_events(ctx->socket_->native_handle(), socket_event::read) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { if (yasio__testbits(ctx->properties_, YCM_TCP)) { @@ -1768,7 +1696,7 @@ void io_service::handle_connect_succeed(transport_handle_t transport) static_cast(transport)->confgure_remote(ctx->remote_eps_[0]); } else - register_descriptor(connection->native_handle(), YEM_POLLIN); + register_descriptor(connection->native_handle(), socket_event::read); if (yasio__testbits(ctx->properties_, YCM_TCP)) { #if defined(SO_NOSIGPIPE) @@ -1845,7 +1773,7 @@ void io_service::handle_connect_failed(io_channel* ctx, int error) YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx->index_, ctx->format_destination().c_str(), error, io_service::strerror(error)); handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx)); } -bool io_service::do_read(transport_handle_t transport) +bool io_service::do_read(transport_handle_t transport, poll_fd_set& revents) { bool ret = false; do @@ -1853,7 +1781,7 @@ bool io_service::do_read(transport_handle_t transport) if (!transport->socket_->is_open()) break; int error = 0; - int revent = pollfd_isset(transport->socket_->native_handle(), POLLIN | POLLHUP | POLLERR | POLLNVAL); + int revent = revents.has_events(transport->socket_->native_handle(), socket_event::read | socket_event::error); int n = transport->do_read(revent, error, this->wait_duration_); if (n >= 0) { @@ -2059,7 +1987,7 @@ bool io_service::cleanup_io(io_base* obj, bool clear_mask) obj->opmask_ = 0; if (obj->socket_->is_open()) { - unregister_descriptor(obj->socket_->native_handle(), YEM_POLLIN | YEM_POLLOUT); + unregister_descriptor(obj->socket_->native_handle(), socket_event::readwrite); obj->socket_->close(); return true; } diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index e5a829a71..d378526a7 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -48,6 +48,7 @@ SOFTWARE. #include "yasio/detail/utils.hpp" #include "yasio/detail/errc.hpp" #include "yasio/detail/byte_buffer.hpp" +#include "yasio/detail/poll_fd_set.hpp" #include "yasio/cxx17/memory.hpp" #include "yasio/cxx17/string_view.hpp" #include "yasio/xxsocket.hpp" @@ -1062,8 +1063,8 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL bool open_internal(io_channel*); - YASIO__DECL void process_transports(); - YASIO__DECL void process_channels(); + YASIO__DECL void process_transports(poll_fd_set& revents); + YASIO__DECL void process_channels(poll_fd_set& revents); YASIO__DECL void process_timers(); YASIO__DECL void interrupt(); @@ -1072,7 +1073,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL int do_resolve(io_channel* ctx); YASIO__DECL void do_connect(io_channel*); - YASIO__DECL void do_connect_completion(io_channel*); + YASIO__DECL void do_connect_completion(io_channel*, poll_fd_set& fd_set); #if defined(YASIO_SSL_BACKEND) YASIO__DECL void init_ssl_context(); @@ -1084,7 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL void process_ares_requests(socket_native_type* socks, int count); + YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, poll_fd_set& revents); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); YASIO__DECL void destroy_ares_channel(); @@ -1100,13 +1101,11 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL void register_descriptor(const socket_native_type fd, int flags); YASIO__DECL void unregister_descriptor(const socket_native_type fd, int flags); - YASIO__DECL void pollfd_mod(socket_native_type fd, int add_flags, int remove_flags, bool use_registry = true); - YASIO__DECL int pollfd_isset(socket_native_type fd, int flags) const; // The major non-blocking event-loop YASIO__DECL void run(void); - YASIO__DECL bool do_read(transport_handle_t); + YASIO__DECL bool do_read(transport_handle_t, poll_fd_set& revents); bool do_write(transport_handle_t transport) { return transport->do_write(this->wait_duration_); } YASIO__DECL void unpack(transport_handle_t, int bytes_expected, int bytes_transferred, int bytes_to_strip); @@ -1129,7 +1128,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // supporting server YASIO__DECL void do_accept(io_channel*); - YASIO__DECL void do_accept_completion(io_channel*); + YASIO__DECL void do_accept_completion(io_channel*, poll_fd_set& revents); YASIO__DECL static const char* strerror(int error); @@ -1172,15 +1171,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // the next wait duration for socket.select highp_time_t wait_duration_; - enum - { - read_op, - write_op, - except_op, - max_ops, - }; - std::vector pollfd_registry_; - std::vector pollfds; + poll_fd_set fd_set_; // options struct __unnamed_options { From 1704539be3990a6e2bf2af92284ecb403cd87ad4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Thu, 29 Dec 2022 21:07:54 +0800 Subject: [PATCH 04/19] Update yasio.cpp --- yasio/yasio.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index fddd0e51e..36855b30e 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -921,7 +921,7 @@ void io_service::run() ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); } #endif - int retval = revents.poll_wait(static_cast(wait_duration / 1000)); + int retval = revents.poll_wait(wait_duration); //YASIO_KLOGV("[core] poll waiting... %ld milliseconds", waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); //int retval = ::poll(this->pollfds.data(), this->pollfds.size(), static_cast(wait_duration / 1000)); //YASIO_KLOGV("[core] poll waked up, retval=%d", retval); From c7967b2aedf43891028b8496967faf9da1dc26be Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 21:11:26 +0800 Subject: [PATCH 05/19] Fix ci --- yasio/yasio.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 36855b30e..916196cd4 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -1211,7 +1211,7 @@ void io_service::do_connect_completion(io_channel* ctx, poll_fd_set& fd_set) { int error = -1; #if !defined(YASIO_SSL_BACKEND) - if (pollfd_isset(ctx->socket_->native_handle(), POLLIN | POLLOUT)) + if (fd_set.has_events(ctx->socket_->native_handle(), socket_event::readwrite)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { From 1a8fc134df819cc24958fb8652aa008f97fd6f18 Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 21:17:03 +0800 Subject: [PATCH 06/19] Fix ci --- yasio/yasio.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 916196cd4..ff085daa1 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -1216,7 +1216,7 @@ void io_service::do_connect_completion(io_channel* ctx, poll_fd_set& fd_set) if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { // The nonblocking tcp handshake complete, remove write event avoid high-CPU occupation - unregister_descriptor(ctx->socket_->native_handle(), YEM_POLLOUT); + unregister_descriptor(ctx->socket_->native_handle(), socket_event::write); handle_connect_succeed(ctx, ctx->socket_); } else From d1c034fe3e4ce6e5828d49b79f1a059e25ddb056 Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 21:29:02 +0800 Subject: [PATCH 07/19] Refactor poll --- yasio/detail/poll_fd_set.hpp | 2 +- yasio/detail/select_fd_set.hpp | 40 +++++++++++++++++++++------------- yasio/yasio.cpp | 5 +---- 3 files changed, 27 insertions(+), 20 deletions(-) diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp index f33b110b7..7b858033a 100644 --- a/yasio/detail/poll_fd_set.hpp +++ b/yasio/detail/poll_fd_set.hpp @@ -25,7 +25,7 @@ class poll_fd_set { return *this; } - int poll_wait(long long timeout_usec) { return ::poll(this->fd_set_.data(), static_cast(this->fd_set_.size()), timeout_usec / 1000); } + int do_poll(long long wait_duration) { return ::poll(this->fd_set_.data(), static_cast(this->fd_set_.size()), wait_duration / 1000); } int has_events(socket_native_type fd, int events) const { diff --git a/yasio/detail/select_fd_set.hpp b/yasio/detail/select_fd_set.hpp index 90c62d87b..441ca0f01 100644 --- a/yasio/detail/select_fd_set.hpp +++ b/yasio/detail/select_fd_set.hpp @@ -19,19 +19,31 @@ namespace inet { class select_fd_set { public: - select_fd_set() { - reset(); + select_fd_set() { reset(); } + + select_fd_set& operator=(select_fd_set& rhs) + { + ::memcpy(this->fd_set_, rhs.fd_set_, sizeof(rhs.fd_set_)); + max_nfds_ = rhs.max_nfds_; + return *this; } - // TODO: cb - int poll_events(int timeout, select_fd_set& revents) + + int do_poll(long long wait_duration) { - // std::copy(this->fd_set_.begin(), this->fd_set_.end(), std::back_inserter(revents.fd_set_)); - // return ::poll(revents, revents.count(), timeout); + timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; + return ::select(this->max_nfds_, &(fd_set_[read_op]), &(fd_set_[write_op]), nullptr, &waitd_tv); } int has_events(socket_native_type fd, int events) const { - + int retval = 0; + if (events & socket_event::read) + retval |= FD_ISSET(fd, &fd_set_[read_op]); + if (events & socket_event::write) + retval |= FD_ISSET(fd, &fd_set_[write_op]); + if (events & socket_event::error) + retval |= FD_ISSET(fd, &fd_set_[except_op]); + return retval; } void reset() @@ -44,13 +56,13 @@ class select_fd_set { void register_descriptor(socket_native_type fd, int events) { - if (yasio__testbits(events, YEM_POLLIN)) + if (yasio__testbits(events, socket_event::read)) FD_SET(fd, &(fd_set_[read_op])); - if (yasio__testbits(events, YEM_POLLOUT)) + if (yasio__testbits(events, socket_event::write)) FD_SET(fd, &(fd_set_[write_op])); - if (yasio__testbits(events, YEM_POLLERR)) + if (yasio__testbits(events, socket_event::error)) FD_SET(fd, &(fd_set_[except_op])); if (max_nfds_ < static_cast(fd) + 1) @@ -59,18 +71,16 @@ class select_fd_set { void deregister_descriptor(socket_native_type fd, int events) { - if (yasio__testbits(events, YEM_POLLIN)) + if (yasio__testbits(events, socket_event::read)) FD_CLR(fd, &(fd_set_[read_op])); - if (yasio__testbits(events, YEM_POLLOUT)) + if (yasio__testbits(events, socket_event::write)) FD_CLR(fd, &(fd_set_[write_op])); - if (yasio__testbits(events, YEM_POLLERR)) + if (yasio__testbits(events, socket_event::error)) FD_CLR(fd, &(fd_set_[except_op])); } -protected: - protected: enum { diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index ff085daa1..e0cb47729 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -921,10 +921,7 @@ void io_service::run() ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); } #endif - int retval = revents.poll_wait(wait_duration); - //YASIO_KLOGV("[core] poll waiting... %ld milliseconds", waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); - //int retval = ::poll(this->pollfds.data(), this->pollfds.size(), static_cast(wait_duration / 1000)); - //YASIO_KLOGV("[core] poll waked up, retval=%d", retval); + int retval = revents.do_poll(wait_duration); if (retval < 0) { int ec = xxsocket::get_last_errno(); From 45d62334d2392bb5c7025ed7450fb523a0b614b3 Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 21:44:48 +0800 Subject: [PATCH 08/19] Refactor code --- yasio/detail/config.hpp | 5 +++++ yasio/detail/fd_set_adapter.hpp | 34 +++++++++++++++++++++++++++++++++ yasio/detail/poll_fd_set.hpp | 2 +- yasio/detail/select_fd_set.hpp | 2 +- yasio/yasio.cpp | 14 +++++++------- yasio/yasio.hpp | 16 ++++++++-------- 6 files changed, 56 insertions(+), 17 deletions(-) create mode 100644 yasio/detail/fd_set_adapter.hpp diff --git a/yasio/detail/config.hpp b/yasio/detail/config.hpp index 281ff64fc..68ab0d4be 100644 --- a/yasio/detail/config.hpp +++ b/yasio/detail/config.hpp @@ -128,6 +128,11 @@ SOFTWARE. */ // #define YASIO_OBS_BUILTIN_STACK 1 +/* +** Uncomment or add compiler flag -DYASIO_DISABLE_POLL to use socket.select for all platforms +*/ +// #define YASIO_DISABLE_POLL 1 + /* ** Workaround for 'vs2013 without full c++11 support', in the future, drop vs2013 support and ** follow 3 lines code will be removed diff --git a/yasio/detail/fd_set_adapter.hpp b/yasio/detail/fd_set_adapter.hpp new file mode 100644 index 000000000..949c6b478 --- /dev/null +++ b/yasio/detail/fd_set_adapter.hpp @@ -0,0 +1,34 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// A multi-platform support c++11 library with focus on asynchronous socket I/O for any +// client application. +////////////////////////////////////////////////////////////////////////////////////////// +// +// detail/fd_set_adapter.hpp +// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) + +// see also: https://github.com/chriskohlhoff/asio +// +#pragma once + +#include "yasio/detail/config.hpp" + +#if !defined(YASIO_DISABLE_POLL) +# include "yasio/detail/poll_fd_set.hpp" +#else +# include "yasio/detail/select_fd_set.hpp" +#endif + +namespace yasio +{ +YASIO__NS_INLINE +namespace inet +{ +#if !defined(YASIO_DISABLE_POLL) +using fd_set_adapter = poll_fd_set; +#else +using fd_set_adapter = select_fd_set; +#endif +} // namespace inet +} // namespace yasio diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp index 7b858033a..9eabaac36 100644 --- a/yasio/detail/poll_fd_set.hpp +++ b/yasio/detail/poll_fd_set.hpp @@ -3,7 +3,7 @@ // client application. ////////////////////////////////////////////////////////////////////////////////////////// // -// detail/poll_event_registry.hpp +// detail/poll_fd_set.hpp // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // // Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) diff --git a/yasio/detail/select_fd_set.hpp b/yasio/detail/select_fd_set.hpp index 441ca0f01..a0b9ebcb9 100644 --- a/yasio/detail/select_fd_set.hpp +++ b/yasio/detail/select_fd_set.hpp @@ -3,7 +3,7 @@ // client application. ////////////////////////////////////////////////////////////////////////////////////////// // -// detail/poll_reactor.hpp +// detail/select_fd_set.hpp // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // // Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index e0cb47729..0ce3fcde8 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -892,7 +892,7 @@ void io_service::run() // The core event loop this->wait_duration_ = YASIO_MAX_WAIT_DURATION; - poll_fd_set revents; // file_descriptor_set + fd_set_adapter revents; // file_descriptor_set do { @@ -965,7 +965,7 @@ void io_service::run() this->state_ = io_service::state::AT_EXITING; } -void io_service::process_transports(poll_fd_set& revents) +void io_service::process_transports(fd_set_adapter& revents) { // preform transports for (auto iter = transports_.begin(); iter != transports_.end();) @@ -987,7 +987,7 @@ void io_service::process_transports(poll_fd_set& revents) iter = transports_.erase(iter); } } -void io_service::process_channels(poll_fd_set& revents) +void io_service::process_channels(fd_set_adapter& revents) { if (!this->channel_ops_.empty()) { @@ -1201,7 +1201,7 @@ void io_service::do_connect(io_channel* ctx) this->handle_connect_failed(ctx, xxsocket::get_last_errno()); } -void io_service::do_connect_completion(io_channel* ctx, poll_fd_set& fd_set) +void io_service::do_connect_completion(io_channel* ctx, fd_set_adapter& fd_set) { assert(ctx->state_ == io_base::state::CONNECTING); if (ctx->state_ == io_base::state::CONNECTING) @@ -1442,7 +1442,7 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -void io_service::process_ares_requests(socket_native_type* socks, int count, poll_fd_set& revents) +void io_service::process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& revents) { if (this->ares_outstanding_work_ > 0) { @@ -1579,7 +1579,7 @@ void io_service::do_accept(io_channel* ctx) handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx, 1)); #endif } -void io_service::do_accept_completion(io_channel* ctx, poll_fd_set& revents) +void io_service::do_accept_completion(io_channel* ctx, fd_set_adapter& revents) { if (ctx->state_ == io_base::state::OPENED) { @@ -1770,7 +1770,7 @@ void io_service::handle_connect_failed(io_channel* ctx, int error) YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx->index_, ctx->format_destination().c_str(), error, io_service::strerror(error)); handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx)); } -bool io_service::do_read(transport_handle_t transport, poll_fd_set& revents) +bool io_service::do_read(transport_handle_t transport, fd_set_adapter& revents) { bool ret = false; do diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index d378526a7..652b05bf8 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -48,7 +48,7 @@ SOFTWARE. #include "yasio/detail/utils.hpp" #include "yasio/detail/errc.hpp" #include "yasio/detail/byte_buffer.hpp" -#include "yasio/detail/poll_fd_set.hpp" +#include "yasio/detail/fd_set_adapter.hpp" #include "yasio/cxx17/memory.hpp" #include "yasio/cxx17/string_view.hpp" #include "yasio/xxsocket.hpp" @@ -1063,8 +1063,8 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL bool open_internal(io_channel*); - YASIO__DECL void process_transports(poll_fd_set& revents); - YASIO__DECL void process_channels(poll_fd_set& revents); + YASIO__DECL void process_transports(fd_set_adapter& revents); + YASIO__DECL void process_channels(fd_set_adapter& revents); YASIO__DECL void process_timers(); YASIO__DECL void interrupt(); @@ -1073,7 +1073,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL int do_resolve(io_channel* ctx); YASIO__DECL void do_connect(io_channel*); - YASIO__DECL void do_connect_completion(io_channel*, poll_fd_set& fd_set); + YASIO__DECL void do_connect_completion(io_channel*, fd_set_adapter& fd_set); #if defined(YASIO_SSL_BACKEND) YASIO__DECL void init_ssl_context(); @@ -1085,7 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, poll_fd_set& revents); + YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& revents); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); YASIO__DECL void destroy_ares_channel(); @@ -1105,7 +1105,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // The major non-blocking event-loop YASIO__DECL void run(void); - YASIO__DECL bool do_read(transport_handle_t, poll_fd_set& revents); + YASIO__DECL bool do_read(transport_handle_t, fd_set_adapter& revents); bool do_write(transport_handle_t transport) { return transport->do_write(this->wait_duration_); } YASIO__DECL void unpack(transport_handle_t, int bytes_expected, int bytes_transferred, int bytes_to_strip); @@ -1128,7 +1128,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // supporting server YASIO__DECL void do_accept(io_channel*); - YASIO__DECL void do_accept_completion(io_channel*, poll_fd_set& revents); + YASIO__DECL void do_accept_completion(io_channel*, fd_set_adapter& revents); YASIO__DECL static const char* strerror(int error); @@ -1171,7 +1171,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // the next wait duration for socket.select highp_time_t wait_duration_; - poll_fd_set fd_set_; + fd_set_adapter fd_set_; // options struct __unnamed_options { From 0eac4affe2eae52fd23aa0d303dc39d3b3d3078e Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 23:09:05 +0800 Subject: [PATCH 09/19] Improve code style --- yasio/detail/poll_fd_set.hpp | 12 +++---- yasio/detail/select_fd_set.hpp | 25 +++++++-------- yasio/yasio.cpp | 57 +++++++++++++++++----------------- yasio/yasio.hpp | 12 +++---- 4 files changed, 52 insertions(+), 54 deletions(-) diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp index 9eabaac36..57f049056 100644 --- a/yasio/detail/poll_fd_set.hpp +++ b/yasio/detail/poll_fd_set.hpp @@ -25,9 +25,11 @@ class poll_fd_set { return *this; } - int do_poll(long long wait_duration) { return ::poll(this->fd_set_.data(), static_cast(this->fd_set_.size()), wait_duration / 1000); } + void reset() { this->fd_set_.clear(); } + + int poll_io(timeval& waitd_tv) { return ::poll(this->fd_set_.data(), static_cast(this->fd_set_.size()), waitd_tv.tv_sec * 1000 + waitd_tv.tv_usec / 1000); } - int has_events(socket_native_type fd, int events) const + int is_set(socket_native_type fd, int events) const { int underlying_events = 0; if (events & socket_event::read) @@ -40,9 +42,7 @@ class poll_fd_set { return it != this->fd_set_.end() ? (it->revents & underlying_events) : 0; } - void reset() { this->fd_set_.clear(); } - - void register_descriptor(socket_native_type fd, int events) + void set(socket_native_type fd, int events) { int underlying_flags = 0; if (yasio__testbits(events, socket_event::read)) @@ -56,7 +56,7 @@ class poll_fd_set { pollfd_mod(this->fd_set_, fd, underlying_flags, 0); } - void deregister_descriptor(socket_native_type fd, int events) + void unset(socket_native_type fd, int events) { int underlying_flags = 0; if (yasio__testbits(events, socket_event::read)) diff --git a/yasio/detail/select_fd_set.hpp b/yasio/detail/select_fd_set.hpp index a0b9ebcb9..a65762e54 100644 --- a/yasio/detail/select_fd_set.hpp +++ b/yasio/detail/select_fd_set.hpp @@ -21,6 +21,14 @@ class select_fd_set { public: select_fd_set() { reset(); } + void reset() + { + FD_ZERO(&fd_set_[read_op]); + FD_ZERO(&fd_set_[write_op]); + FD_ZERO(&fd_set_[except_op]); + max_nfds_ = 0; + } + select_fd_set& operator=(select_fd_set& rhs) { ::memcpy(this->fd_set_, rhs.fd_set_, sizeof(rhs.fd_set_)); @@ -28,13 +36,12 @@ class select_fd_set { return *this; } - int do_poll(long long wait_duration) + int poll_io(timeval& waitd_tv) { - timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; return ::select(this->max_nfds_, &(fd_set_[read_op]), &(fd_set_[write_op]), nullptr, &waitd_tv); } - int has_events(socket_native_type fd, int events) const + int is_set(socket_native_type fd, int events) const { int retval = 0; if (events & socket_event::read) @@ -46,15 +53,7 @@ class select_fd_set { return retval; } - void reset() - { - FD_ZERO(&fd_set_[read_op]); - FD_ZERO(&fd_set_[write_op]); - FD_ZERO(&fd_set_[except_op]); - max_nfds_ = 0; - } - - void register_descriptor(socket_native_type fd, int events) + void set(socket_native_type fd, int events) { if (yasio__testbits(events, socket_event::read)) FD_SET(fd, &(fd_set_[read_op])); @@ -69,7 +68,7 @@ class select_fd_set { max_nfds_ = static_cast(fd) + 1; } - void deregister_descriptor(socket_native_type fd, int events) + void unset(socket_native_type fd, int events) { if (yasio__testbits(events, socket_event::read)) FD_CLR(fd, &(fd_set_[read_op])); diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 0ce3fcde8..e03ea6e7f 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -384,7 +384,7 @@ bool io_transport::do_write(highp_time_t& wait_duration) } if (no_wevent && pollout_registerred_) { - get_service().unregister_descriptor(socket_->native_handle(), socket_event::write); + get_service().deregister_descriptor(socket_->native_handle(), socket_event::write); pollout_registerred_ = false; } ret = true; @@ -825,7 +825,7 @@ void io_service::finalize() life_token_.reset(); #endif destroy_channels(); - unregister_descriptor(interrupter_.read_descriptor(), socket_event::read); + deregister_descriptor(interrupter_.read_descriptor(), socket_event::read); options_.on_event_ = nullptr; options_.resolv_ = nullptr; @@ -890,17 +890,16 @@ void io_service::run() this->ipsv_ = static_cast(xxsocket::getipsv()); // The core event loop + fd_set_adapter fd_set; // file_descriptor_set this->wait_duration_ = YASIO_MAX_WAIT_DURATION; - fd_set_adapter revents; // file_descriptor_set - do { auto wait_duration = get_timeout(this->wait_duration_); // Gets current wait duration this->wait_duration_ = YASIO_MAX_WAIT_DURATION; // Reset next wait duration if (wait_duration > 0) { - revents = this->fd_set_; + fd_set = this->fd_set_; timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; #if defined(YASIO_HAVE_CARES) if (this->ares_outstanding_work_ > 0) @@ -913,7 +912,7 @@ void io_service::run() { auto fd = ares_socks[i]; ++ares_socks_count; - revents.register_descriptor(fd, socket_event::readwrite); + fd_set.set(fd, socket_event::readwrite); } else break; @@ -921,7 +920,7 @@ void io_service::run() ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); } #endif - int retval = revents.do_poll(wait_duration); + int retval = fd_set.poll_io(waitd_tv); if (retval < 0) { int ec = xxsocket::get_last_errno(); @@ -933,7 +932,7 @@ void io_service::run() if (retval == 0) YASIO_KLOGV("[core] %s", "do_select is timeout, process_timers()"); - else if (revents.has_events(this->interrupter_.read_descriptor(), socket_event::read)) + else if (fd_set.is_set(this->interrupter_.read_descriptor(), socket_event::read)) { // Reset the interrupter. if (!interrupter_.reset()) interrupter_.recreate(); @@ -943,14 +942,14 @@ void io_service::run() #if defined(YASIO_HAVE_CARES) // process possible async resolve requests. - process_ares_requests(ares_socks, ares_socks_count, revents); + process_ares_requests(ares_socks, ares_socks_count, fd_set); #endif // process active transports - process_transports(revents); + process_transports(fd_set); // process active channels - process_channels(revents); + process_channels(fd_set); // process timeout timers process_timers(); @@ -965,13 +964,13 @@ void io_service::run() this->state_ = io_service::state::AT_EXITING; } -void io_service::process_transports(fd_set_adapter& revents) +void io_service::process_transports(fd_set_adapter& fd_set) { // preform transports for (auto iter = transports_.begin(); iter != transports_.end();) { auto transport = *iter; - bool ok = (do_read(transport, revents) && do_write(transport)); + bool ok = (do_read(transport, fd_set) && do_write(transport)); if (ok) { int opm = transport->opmask_ | transport->ctx_->opmask_ | this->stop_flag_; @@ -987,7 +986,7 @@ void io_service::process_transports(fd_set_adapter& revents) iter = transports_.erase(iter); } } -void io_service::process_channels(fd_set_adapter& revents) +void io_service::process_channels(fd_set_adapter& fd_set) { if (!this->channel_ops_.empty()) { @@ -1013,7 +1012,7 @@ void io_service::process_channels(fd_set_adapter& revents) handle_connect_failed(ctx, ctx->error_); } else if (ctx->state_ == io_base::state::CONNECTING) - do_connect_completion(ctx, revents); + do_connect_completion(ctx, fd_set); finish = ctx->error_ != EINPROGRESS; } else if (yasio__testbits(ctx->properties_, YCM_SERVER)) @@ -1026,7 +1025,7 @@ void io_service::process_channels(fd_set_adapter& revents) finish = (ctx->state_ != io_base::state::OPENED); if (!finish) - do_accept_completion(ctx, revents); + do_accept_completion(ctx, fd_set); else ctx->bytes_transferred_ = 0; } @@ -1101,8 +1100,8 @@ void io_service::handle_close(transport_handle_t thandle) cleanup_channel(ctx, false); } } -void io_service::register_descriptor(const socket_native_type fd, int flags) { this->fd_set_.register_descriptor(fd, flags); } -void io_service::unregister_descriptor(const socket_native_type fd, int flags) { this->fd_set_.deregister_descriptor(fd, flags); } +void io_service::register_descriptor(const socket_native_type fd, int flags) { this->fd_set_.set(fd, flags); } +void io_service::deregister_descriptor(const socket_native_type fd, int flags) { this->fd_set_.unset(fd, flags); } int io_service::write(transport_handle_t transport, dynamic_buffer_t buffer, completion_cb_t handler) { @@ -1208,12 +1207,12 @@ void io_service::do_connect_completion(io_channel* ctx, fd_set_adapter& fd_set) { int error = -1; #if !defined(YASIO_SSL_BACKEND) - if (fd_set.has_events(ctx->socket_->native_handle(), socket_event::readwrite)) + if (fd_set.is_set(ctx->socket_->native_handle(), socket_event::readwrite)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { // The nonblocking tcp handshake complete, remove write event avoid high-CPU occupation - unregister_descriptor(ctx->socket_->native_handle(), socket_event::write); + deregister_descriptor(ctx->socket_->native_handle(), socket_event::write); handle_connect_succeed(ctx, ctx->socket_); } else @@ -1223,12 +1222,12 @@ void io_service::do_connect_completion(io_channel* ctx, fd_set_adapter& fd_set) #else if (!yasio__testbits(ctx->properties_, YCPF_SSL_HANDSHAKING)) { - if (fd_set.has_events(ctx->socket_->native_handle(), socket_event::readwrite)) + if (fd_set.is_set(ctx->socket_->native_handle(), socket_event::readwrite)) { if (ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { // The nonblocking tcp handshake complete, remove write event avoid high-CPU occupation - unregister_descriptor(ctx->socket_->native_handle(), socket_event::write); + deregister_descriptor(ctx->socket_->native_handle(), socket_event::write); if (!yasio__testbits(ctx->properties_, YCM_SSL)) handle_connect_succeed(ctx, ctx->socket_); else @@ -1442,14 +1441,14 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -void io_service::process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& revents) +void io_service::process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set) { if (this->ares_outstanding_work_ > 0) { for (auto i = 0; i < count; ++i) { auto fd = socks[i]; - ::ares_process_fd(this->ares_, revents.has_events(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, revents.has_events(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); + ::ares_process_fd(this->ares_, fd_set.is_set(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, fd_set.is_set(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); } } } @@ -1579,12 +1578,12 @@ void io_service::do_accept(io_channel* ctx) handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx, 1)); #endif } -void io_service::do_accept_completion(io_channel* ctx, fd_set_adapter& revents) +void io_service::do_accept_completion(io_channel* ctx, fd_set_adapter& fd_set) { if (ctx->state_ == io_base::state::OPENED) { int error = 0; - if (revents.has_events(ctx->socket_->native_handle(), socket_event::read) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) + if (fd_set.is_set(ctx->socket_->native_handle(), socket_event::read) && ctx->socket_->get_optval(SOL_SOCKET, SO_ERROR, error) >= 0 && error == 0) { if (yasio__testbits(ctx->properties_, YCM_TCP)) { @@ -1770,7 +1769,7 @@ void io_service::handle_connect_failed(io_channel* ctx, int error) YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx->index_, ctx->format_destination().c_str(), error, io_service::strerror(error)); handle_event(cxx14::make_unique(ctx->index_, YEK_ON_OPEN, error, ctx)); } -bool io_service::do_read(transport_handle_t transport, fd_set_adapter& revents) +bool io_service::do_read(transport_handle_t transport, fd_set_adapter& fd_set) { bool ret = false; do @@ -1778,7 +1777,7 @@ bool io_service::do_read(transport_handle_t transport, fd_set_adapter& revents) if (!transport->socket_->is_open()) break; int error = 0; - int revent = revents.has_events(transport->socket_->native_handle(), socket_event::read | socket_event::error); + int revent = fd_set.is_set(transport->socket_->native_handle(), socket_event::read | socket_event::error); int n = transport->do_read(revent, error, this->wait_duration_); if (n >= 0) { @@ -1984,7 +1983,7 @@ bool io_service::cleanup_io(io_base* obj, bool clear_mask) obj->opmask_ = 0; if (obj->socket_->is_open()) { - unregister_descriptor(obj->socket_->native_handle(), socket_event::readwrite); + deregister_descriptor(obj->socket_->native_handle(), socket_event::readwrite); obj->socket_->close(); return true; } diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 652b05bf8..565023d17 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1063,8 +1063,8 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL bool open_internal(io_channel*); - YASIO__DECL void process_transports(fd_set_adapter& revents); - YASIO__DECL void process_channels(fd_set_adapter& revents); + YASIO__DECL void process_transports(fd_set_adapter& fd_set); + YASIO__DECL void process_channels(fd_set_adapter& fd_set); YASIO__DECL void process_timers(); YASIO__DECL void interrupt(); @@ -1085,7 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& revents); + YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); YASIO__DECL void destroy_ares_channel(); @@ -1100,12 +1100,12 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL void deallocate_transport(transport_handle_t); YASIO__DECL void register_descriptor(const socket_native_type fd, int flags); - YASIO__DECL void unregister_descriptor(const socket_native_type fd, int flags); + YASIO__DECL void deregister_descriptor(const socket_native_type fd, int flags); // The major non-blocking event-loop YASIO__DECL void run(void); - YASIO__DECL bool do_read(transport_handle_t, fd_set_adapter& revents); + YASIO__DECL bool do_read(transport_handle_t, fd_set_adapter& fd_set); bool do_write(transport_handle_t transport) { return transport->do_write(this->wait_duration_); } YASIO__DECL void unpack(transport_handle_t, int bytes_expected, int bytes_transferred, int bytes_to_strip); @@ -1128,7 +1128,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] // supporting server YASIO__DECL void do_accept(io_channel*); - YASIO__DECL void do_accept_completion(io_channel*, fd_set_adapter& revents); + YASIO__DECL void do_accept_completion(io_channel*, fd_set_adapter& fd_set); YASIO__DECL static const char* strerror(int error); From 807ee2d9904d97ac5a6bbb74e8dd102c137f50ec Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 23:11:51 +0800 Subject: [PATCH 10/19] Tidy code [skip ci] --- yasio/detail/fd_set_adapter.hpp | 2 -- yasio/detail/poll_fd_set.hpp | 1 - 2 files changed, 3 deletions(-) diff --git a/yasio/detail/fd_set_adapter.hpp b/yasio/detail/fd_set_adapter.hpp index 949c6b478..6bef1163b 100644 --- a/yasio/detail/fd_set_adapter.hpp +++ b/yasio/detail/fd_set_adapter.hpp @@ -8,8 +8,6 @@ // // Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) -// see also: https://github.com/chriskohlhoff/asio -// #pragma once #include "yasio/detail/config.hpp" diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp index 57f049056..5c6a3e53c 100644 --- a/yasio/detail/poll_fd_set.hpp +++ b/yasio/detail/poll_fd_set.hpp @@ -7,7 +7,6 @@ // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // // Copyright (c) 2012-2022 HALX99 (halx99 at live dot com) - #pragma once #include #include "yasio/detail/socket.hpp" From 9eb5b9f0b27d90323b8aef85ecf50b0bad523bb0 Mon Sep 17 00:00:00 2001 From: halx99 Date: Thu, 29 Dec 2022 23:26:47 +0800 Subject: [PATCH 11/19] Improve code style --- yasio/yasio.cpp | 47 ++++++++++++++++++++++++++--------------------- yasio/yasio.hpp | 1 + 2 files changed, 27 insertions(+), 21 deletions(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index e03ea6e7f..8961b2996 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -890,35 +890,18 @@ void io_service::run() this->ipsv_ = static_cast(xxsocket::getipsv()); // The core event loop - fd_set_adapter fd_set; // file_descriptor_set + fd_set_adapter fd_set; // The temp file descriptor set this->wait_duration_ = YASIO_MAX_WAIT_DURATION; - do { auto wait_duration = get_timeout(this->wait_duration_); // Gets current wait duration this->wait_duration_ = YASIO_MAX_WAIT_DURATION; // Reset next wait duration if (wait_duration > 0) { - fd_set = this->fd_set_; + fd_set = this->fd_set_; timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; #if defined(YASIO_HAVE_CARES) - if (this->ares_outstanding_work_ > 0) - { - int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); - ares_socks_count = 0; - for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) - { - if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) - { - auto fd = ares_socks[i]; - ++ares_socks_count; - fd_set.set(fd, socket_event::readwrite); - } - else - break; - } - ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); - } + ares_socks_count = ares_set_fds(ares_socks, fd_set, waitd_tv); #endif int retval = fd_set.poll_io(waitd_tv); if (retval < 0) @@ -1441,6 +1424,27 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } +int io_service::ares_set_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set, timeval& waitd_tv) +{ + int count = 0; + if (this->ares_outstanding_work_ > 0) + { + int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); + for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) + { + if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) + { + auto fd = ares_socks[i]; + ++count; + fd_set.set(fd, socket_event::readwrite); + } + else + break; + } + ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); + } + return count; +} void io_service::process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set) { if (this->ares_outstanding_work_ > 0) @@ -1448,7 +1452,8 @@ void io_service::process_ares_requests(socket_native_type* socks, int count, fd_ for (auto i = 0; i < count; ++i) { auto fd = socks[i]; - ::ares_process_fd(this->ares_, fd_set.is_set(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, fd_set.is_set(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); + ::ares_process_fd(this->ares_, fd_set.is_set(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, + fd_set.is_set(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); } } } diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 565023d17..729ce0f7b 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1085,6 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); + YASIO__DECL int ares_set_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set, timeval& waitd_tv); YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); From 662564224ea1b7652d3b9abc1741803c8d89f7f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 08:46:27 +0800 Subject: [PATCH 12/19] Improve code style --- yasio/detail/poll_fd_set.hpp | 33 ++++++++++++++++----------------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/yasio/detail/poll_fd_set.hpp b/yasio/detail/poll_fd_set.hpp index 5c6a3e53c..f52472ed4 100644 --- a/yasio/detail/poll_fd_set.hpp +++ b/yasio/detail/poll_fd_set.hpp @@ -43,50 +43,49 @@ class poll_fd_set { void set(socket_native_type fd, int events) { - int underlying_flags = 0; + int underlying_events = 0; if (yasio__testbits(events, socket_event::read)) - underlying_flags |= POLLIN; + underlying_events |= POLLIN; if (yasio__testbits(events, socket_event::write)) - underlying_flags |= POLLOUT; + underlying_events |= POLLOUT; if (yasio__testbits(events, socket_event::error)) - underlying_flags |= POLLERR; - pollfd_mod(this->fd_set_, fd, underlying_flags, 0); + underlying_events |= POLLERR; + pollfd_mod(this->fd_set_, fd, underlying_events, 0); } void unset(socket_native_type fd, int events) { - int underlying_flags = 0; + int underlying_events = 0; if (yasio__testbits(events, socket_event::read)) - underlying_flags |= POLLIN; + underlying_events |= POLLIN; if (yasio__testbits(events, socket_event::write)) - underlying_flags |= POLLOUT; + underlying_events |= POLLOUT; if (yasio__testbits(events, socket_event::error)) - underlying_flags |= POLLERR; + underlying_events |= POLLERR; - pollfd_mod(this->fd_set_, fd, 0, underlying_flags); + pollfd_mod(this->fd_set_, fd, 0, underlying_events); } protected: - static void pollfd_mod(std::vector& fdset, socket_native_type fd, int add_flags, int remove_flags) + static void pollfd_mod(std::vector& fdset, socket_native_type fd, int add_events, int remove_events) { auto it = std::find_if(fdset.begin(), fdset.end(), [fd](const pollfd& pfd) { return pfd.fd == fd; }); if (it != fdset.end()) { - // POLLIN - it->events |= add_flags; - it->events &= ~remove_flags; + it->events |= add_events; + it->events &= ~remove_events; if (it->events == 0) fdset.erase(it); } else { - auto combined_flags = add_flags & ~remove_flags; - if (combined_flags) - fdset.push_back(pollfd{fd, static_cast(combined_flags), 0}); + auto events = add_events & ~remove_events; + if (events) + fdset.push_back(pollfd{fd, static_cast(events), 0}); } } From 0e92aca213863a5f4ca32bc9e88f794f3a667f6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 08:49:59 +0800 Subject: [PATCH 13/19] Update yasio.hpp --- yasio/yasio.hpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 729ce0f7b..8fe017d3a 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1100,8 +1100,8 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL transport_handle_t allocate_transport(io_channel*, xxsocket_ptr&&); YASIO__DECL void deallocate_transport(transport_handle_t); - YASIO__DECL void register_descriptor(const socket_native_type fd, int flags); - YASIO__DECL void deregister_descriptor(const socket_native_type fd, int flags); + YASIO__DECL void register_descriptor(const socket_native_type fd, int events); + YASIO__DECL void deregister_descriptor(const socket_native_type fd, int events); // The major non-blocking event-loop YASIO__DECL void run(void); From 97c7994a633b0cf3bb388c6c0709c208e066cf53 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 08:51:27 +0800 Subject: [PATCH 14/19] Update yasio.cpp --- yasio/yasio.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 8961b2996..985dfeb4f 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -1083,8 +1083,8 @@ void io_service::handle_close(transport_handle_t thandle) cleanup_channel(ctx, false); } } -void io_service::register_descriptor(const socket_native_type fd, int flags) { this->fd_set_.set(fd, flags); } -void io_service::deregister_descriptor(const socket_native_type fd, int flags) { this->fd_set_.unset(fd, flags); } +void io_service::register_descriptor(const socket_native_type fd, int events) { this->fd_set_.set(fd, events); } +void io_service::deregister_descriptor(const socket_native_type fd, int events) { this->fd_set_.unset(fd, events); } int io_service::write(transport_handle_t transport, dynamic_buffer_t buffer, completion_cb_t handler) { From 2734f871202bf3798ba655634529dabc60bc93aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 08:57:33 +0800 Subject: [PATCH 15/19] Update config.hpp --- yasio/detail/config.hpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/yasio/detail/config.hpp b/yasio/detail/config.hpp index 68ab0d4be..a9288e73d 100644 --- a/yasio/detail/config.hpp +++ b/yasio/detail/config.hpp @@ -97,7 +97,7 @@ SOFTWARE. // #define YASIO_ENABLE_UDS 1 /* -** Uncomment or add compiler flag -DYASIO_NT_COMPAT_GAI for earlier versions of Windows XP +** Uncomment or add compiler flag -DYASIO_NT_COMPAT_GAI for compatible with Windows XP ** see: https://docs.microsoft.com/en-us/windows/win32/api/ws2tcpip/nf-ws2tcpip-getaddrinfo */ // #define YASIO_NT_COMPAT_GAI 1 @@ -130,6 +130,7 @@ SOFTWARE. /* ** Uncomment or add compiler flag -DYASIO_DISABLE_POLL to use socket.select for all platforms +** If you need support Windows XP, you need disable poll */ // #define YASIO_DISABLE_POLL 1 From 58be357b6252069d14af0a90ac810a5e86dc3288 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 08:59:44 +0800 Subject: [PATCH 16/19] Update CMakeLists.txt --- CMakeLists.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 3e8ecc851..9942237d4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -401,6 +401,7 @@ macro (yasio_config_lib_options target_name) yasio_config_pred(${target_name} YASIO_HAVE_HALF_FLOAT) yasio_config_pred(${target_name} YASIO_ENABLE_PASSIVE_EVENT) yasio_config_pred(${target_name} YASIO_NO_JNI_ONLOAD) + yasio_config_pred(${target_name} YASIO_DISABLE_POLL) endmacro() yasio_config_lib_options(${yasio_target_name}) From f7e7832d82e688f6ee2de7cd69a968b4f8323acc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=BA=BF=E7=81=B5=7CDeal?= Date: Fri, 30 Dec 2022 09:23:24 +0800 Subject: [PATCH 17/19] Update CHANGELOG.md[skip ci] --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1b9659bb..6e54941b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,10 @@ +yasio-3.39.6 + + 1. Reimplement reactor backends: select, poll + 2. Now the default reactor backend is poll + 3. Add compiler flag `YASIO_DISABLE_POLL` for switching to socket.select as same with previous releases + + yasio-3.39.5 1. Fix no callback when resolve domain failed by `getaddrinfo` From 79445c4d89c66d71c657df419830a5fa52c8a121 Mon Sep 17 00:00:00 2001 From: halx99 Date: Fri, 30 Dec 2022 10:55:48 +0800 Subject: [PATCH 18/19] Improve code style [skip ci] --- yasio/yasio.cpp | 27 +++++++++++++-------------- yasio/yasio.hpp | 2 +- 2 files changed, 14 insertions(+), 15 deletions(-) diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 8961b2996..d32ddf69a 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -901,7 +901,10 @@ void io_service::run() fd_set = this->fd_set_; timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; #if defined(YASIO_HAVE_CARES) - ares_socks_count = ares_set_fds(ares_socks, fd_set, waitd_tv); + if (ares_outstanding_work_) { + ares_socks_count = set_ares_fds(ares_socks, fd_set); + ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); + } #endif int retval = fd_set.poll_io(waitd_tv); if (retval < 0) @@ -1424,24 +1427,20 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -int io_service::ares_set_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set, timeval& waitd_tv) +int io_service::set_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set) { int count = 0; - if (this->ares_outstanding_work_ > 0) + int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); + for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) { - int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); - for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) + if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) { - if (ARES_GETSOCK_READABLE(bitmask, i) || ARES_GETSOCK_WRITABLE(bitmask, i)) - { - auto fd = ares_socks[i]; - ++count; - fd_set.set(fd, socket_event::readwrite); - } - else - break; + auto fd = ares_socks[i]; + ++count; + fd_set.set(fd, socket_event::readwrite); } - ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); + else + break; } return count; } diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 729ce0f7b..2f840ab82 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1085,7 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL int ares_set_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set, timeval& waitd_tv); + YASIO__DECL int set_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set); YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers(); From 2e3fdbb5a70c99a006e04ef33d03e1b6e667d205 Mon Sep 17 00:00:00 2001 From: halx99 Date: Fri, 30 Dec 2022 12:01:09 +0800 Subject: [PATCH 19/19] Improve code style [skip ci] --- yasio/detail/config.hpp | 10 ---------- yasio/detail/socket.hpp | 5 +++++ yasio/yasio.cpp | 4 ++-- yasio/yasio.hpp | 2 +- 4 files changed, 8 insertions(+), 13 deletions(-) diff --git a/yasio/detail/config.hpp b/yasio/detail/config.hpp index a9288e73d..955a9b9b7 100644 --- a/yasio/detail/config.hpp +++ b/yasio/detail/config.hpp @@ -238,14 +238,4 @@ SOFTWARE. #define YASIO_SSL_PIN "yasio_ssl_client" #define YASIO_SSL_PIN_LEN (sizeof(YASIO_SSL_PIN) - 1) - -/* -** yasio bitop macros -*/ - -#define yasio__setbits(x, m) ((x) |= (m)) -#define yasio__clearbits(x, m) ((x) &= ~(m)) -#define yasio__testbits(x, m) ((x) & (m)) -#define yasio__setlobyte(x, v) ((x) = ((x) & ~((decltype(x))0xff)) | (v)) - #endif diff --git a/yasio/detail/socket.hpp b/yasio/detail/socket.hpp index d682b3974..d3014b3ba 100644 --- a/yasio/detail/socket.hpp +++ b/yasio/detail/socket.hpp @@ -228,6 +228,11 @@ inline bool IN6_IS_ADDR_GLOBAL(const in6_addr* a) #define YASIO_ADDR_ANY(af) (af == AF_INET ? "0.0.0.0" : "::") +#define yasio__setbits(x, m) ((x) |= (m)) +#define yasio__clearbits(x, m) ((x) &= ~(m)) +#define yasio__testbits(x, m) ((x) & (m)) +#define yasio__setlobyte(x, v) ((x) = ((x) & ~((decltype(x))0xff)) | (v)) + namespace yasio { YASIO__NS_INLINE diff --git a/yasio/yasio.cpp b/yasio/yasio.cpp index 673133fed..888521b3c 100644 --- a/yasio/yasio.cpp +++ b/yasio/yasio.cpp @@ -902,7 +902,7 @@ void io_service::run() timeval waitd_tv = {(decltype(timeval::tv_sec))(wait_duration / 1000000), (decltype(timeval::tv_usec))(wait_duration % 1000000)}; #if defined(YASIO_HAVE_CARES) if (ares_outstanding_work_) { - ares_socks_count = set_ares_fds(ares_socks, fd_set); + ares_socks_count = register_ares_fds(ares_socks, fd_set); ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); } #endif @@ -1427,7 +1427,7 @@ void io_service::ares_getaddrinfo_cb(void* arg, int status, int /*timeouts*/, ar } current_service.interrupt(); } -int io_service::set_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set) +int io_service::register_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set) { int count = 0; int bitmask = ::ares_getsock(this->ares_, ares_socks, ARES_GETSOCK_MAXNUM); diff --git a/yasio/yasio.hpp b/yasio/yasio.hpp index 18b7d8cb5..cd281418e 100644 --- a/yasio/yasio.hpp +++ b/yasio/yasio.hpp @@ -1085,7 +1085,7 @@ class YASIO_API io_service // lgtm [cpp/class-many-fields] YASIO__DECL static void ares_getaddrinfo_cb(void* arg, int status, int timeouts, ares_addrinfo* answerlist); YASIO__DECL void ares_work_started(); YASIO__DECL void ares_work_finished(); - YASIO__DECL int set_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set); + YASIO__DECL int register_ares_fds(socket_native_type* ares_socks, fd_set_adapter& fd_set); YASIO__DECL void process_ares_requests(socket_native_type* socks, int count, fd_set_adapter& fd_set); YASIO__DECL void recreate_ares_channel(); YASIO__DECL void config_ares_name_servers();