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` 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}) diff --git a/yasio/detail/config.hpp b/yasio/detail/config.hpp index ce2da5563..955a9b9b7 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 @@ -128,6 +128,12 @@ SOFTWARE. */ // #define YASIO_OBS_BUILTIN_STACK 1 +/* +** 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 + /* ** Workaround for 'vs2013 without full c++11 support', in the future, drop vs2013 support and ** follow 3 lines code will be removed @@ -192,7 +198,7 @@ SOFTWARE. /* ** The yasio version macros */ -#define YASIO_VERSION_NUM 0x033905 +#define YASIO_VERSION_NUM 0x033906 /* ** The macros used by io_service. diff --git a/yasio/detail/fd_set_adapter.hpp b/yasio/detail/fd_set_adapter.hpp new file mode 100644 index 000000000..6bef1163b --- /dev/null +++ b/yasio/detail/fd_set_adapter.hpp @@ -0,0 +1,32 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// 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) + +#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 new file mode 100644 index 000000000..f52472ed4 --- /dev/null +++ b/yasio/detail/poll_fd_set.hpp @@ -0,0 +1,96 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// A multi-platform support c++11 library with focus on asynchronous socket I/O for any +// client application. +////////////////////////////////////////////////////////////////////////////////////////// +// +// detail/poll_fd_set.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; + } + + 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 is_set(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 set(socket_native_type fd, int events) + { + int underlying_events = 0; + if (yasio__testbits(events, socket_event::read)) + underlying_events |= POLLIN; + + if (yasio__testbits(events, socket_event::write)) + underlying_events |= POLLOUT; + + if (yasio__testbits(events, socket_event::error)) + underlying_events |= POLLERR; + pollfd_mod(this->fd_set_, fd, underlying_events, 0); + } + + void unset(socket_native_type fd, int events) + { + int underlying_events = 0; + if (yasio__testbits(events, socket_event::read)) + underlying_events |= POLLIN; + + if (yasio__testbits(events, socket_event::write)) + underlying_events |= POLLOUT; + + if (yasio__testbits(events, socket_event::error)) + underlying_events |= POLLERR; + + pollfd_mod(this->fd_set_, fd, 0, underlying_events); + } + +protected: + 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()) + { + it->events |= add_events; + it->events &= ~remove_events; + if (it->events == 0) + fdset.erase(it); + } + else + { + auto events = add_events & ~remove_events; + if (events) + fdset.push_back(pollfd{fd, static_cast(events), 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..a65762e54 --- /dev/null +++ b/yasio/detail/select_fd_set.hpp @@ -0,0 +1,95 @@ +////////////////////////////////////////////////////////////////////////////////////////// +// A multi-platform support c++11 library with focus on asynchronous socket I/O for any +// client application. +////////////////////////////////////////////////////////////////////////////////////////// +// +// detail/select_fd_set.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(); } + + 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_)); + max_nfds_ = rhs.max_nfds_; + return *this; + } + + int poll_io(timeval& waitd_tv) + { + return ::select(this->max_nfds_, &(fd_set_[read_op]), &(fd_set_[write_op]), nullptr, &waitd_tv); + } + + int is_set(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 set(socket_native_type fd, int events) + { + if (yasio__testbits(events, socket_event::read)) + FD_SET(fd, &(fd_set_[read_op])); + + if (yasio__testbits(events, socket_event::write)) + FD_SET(fd, &(fd_set_[write_op])); + + if (yasio__testbits(events, socket_event::error)) + FD_SET(fd, &(fd_set_[except_op])); + + if (max_nfds_ < static_cast(fd) + 1) + max_nfds_ = static_cast(fd) + 1; + } + + void unset(socket_native_type fd, int events) + { + if (yasio__testbits(events, socket_event::read)) + FD_CLR(fd, &(fd_set_[read_op])); + + if (yasio__testbits(events, socket_event::write)) + FD_CLR(fd, &(fd_set_[write_op])); + + if (yasio__testbits(events, socket_event::error)) + FD_CLR(fd, &(fd_set_[except_op])); + } + +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..d3014b3ba 100644 --- a/yasio/detail/socket.hpp +++ b/yasio/detail/socket.hpp @@ -228,4 +228,25 @@ 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 +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 a66c6436b..888521b3c 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 @@ -386,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; } } @@ -395,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().deregister_descriptor(socket_->native_handle(), socket_event::write); pollout_registerred_ = false; } ret = true; @@ -815,13 +804,8 @@ 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); + register_descriptor(interrupter_.read_descriptor(), socket_event::read); // create channels create_channels(channel_eps, channel_count); @@ -841,7 +825,7 @@ void io_service::finalize() life_token_.reset(); #endif destroy_channels(); - unregister_descriptor(interrupter_.read_descriptor(), YEM_POLLIN); + deregister_descriptor(interrupter_.read_descriptor(), socket_event::read); options_.on_event_ = nullptr; options_.resolv_ = nullptr; @@ -898,13 +882,15 @@ 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]; + fd_set_adapter fd_set; // The temp file descriptor set this->wait_duration_ = YASIO_MAX_WAIT_DURATION; do { @@ -912,7 +898,15 @@ 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); + 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 (ares_outstanding_work_) { + ares_socks_count = register_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) { int ec = xxsocket::get_last_errno(); @@ -924,7 +918,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 (fd_set.is_set(this->interrupter_.read_descriptor(), socket_event::read)) { // Reset the interrupter. if (!interrupter_.reset()) interrupter_.recreate(); @@ -934,14 +928,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, fd_set); #endif // process active transports - process_transports(fds_array); + process_transports(fd_set); // process active channels - process_channels(fds_array); + process_channels(fd_set); // process timeout timers process_timers(); @@ -956,13 +950,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(fd_set_adapter& fd_set) { // 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, fd_set) && do_write(transport)); if (ok) { int opm = transport->opmask_ | transport->ctx_->opmask_ | this->stop_flag_; @@ -978,7 +972,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(fd_set_adapter& fd_set) { if (!this->channel_ops_.empty()) { @@ -1004,7 +998,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, fd_set); finish = ctx->error_ != EINPROGRESS; } else if (yasio__testbits(ctx->properties_, YCM_SERVER)) @@ -1017,7 +1011,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, fd_set); else ctx->bytes_transferred_ = 0; } @@ -1092,31 +1086,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) -{ - if (yasio__testbits(flags, YEM_POLLIN)) - FD_SET(fd, &(fds_array_[read_op])); - - if (yasio__testbits(flags, YEM_POLLOUT)) - FD_SET(fd, &(fds_array_[write_op])); +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); } - 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; -} -void io_service::unregister_descriptor(const socket_native_type fd, int flags) -{ - if (yasio__testbits(flags, YEM_POLLIN)) - FD_CLR(fd, &(fds_array_[read_op])); - - if (yasio__testbits(flags, YEM_POLLOUT)) - FD_CLR(fd, &(fds_array_[write_op])); - - if (yasio__testbits(flags, YEM_POLLERR)) - FD_CLR(fd, &(fds_array_[except_op])); -} int io_service::write(transport_handle_t transport, dynamic_buffer_t buffer, completion_cb_t handler) { if (transport && transport->is_open()) @@ -1196,7 +1168,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) @@ -1206,7 +1178,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 } @@ -1214,19 +1186,19 @@ 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, fd_set_adapter& fd_set) { assert(ctx->state_ == io_base::state::CONNECTING); if (ctx->state_ == io_base::state::CONNECTING) { 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 (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(), YEM_POLLOUT); + deregister_descriptor(ctx->socket_->native_handle(), socket_event::write); handle_connect_succeed(ctx, ctx->socket_); } else @@ -1236,12 +1208,12 @@ 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 (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(), YEM_POLLOUT); + deregister_descriptor(ctx->socket_->native_handle(), socket_event::write); if (!yasio__testbits(ctx->properties_, YCM_SSL)) handle_connect_succeed(ctx, ctx->socket_); else @@ -1455,22 +1427,32 @@ 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) +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); + 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; + } + 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) { - 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_, fd_set.is_set(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, + fd_set.is_set(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); } } } @@ -1584,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); @@ -1600,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, fd_set* fds_array) +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 (FD_ISSET(ctx->socket_->native_handle(), &fds_array[read_op]) && 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)) { @@ -1714,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) @@ -1791,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, fd_set* fds_array) +bool io_service::do_read(transport_handle_t transport, fd_set_adapter& fd_set) { bool ret = false; do @@ -1799,7 +1781,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 = 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) { @@ -1969,24 +1951,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()) @@ -2023,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); + 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 86a90ffea..cd281418e 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/fd_set_adapter.hpp" #include "yasio/cxx17/memory.hpp" #include "yasio/cxx17/string_view.hpp" #include "yasio/xxsocket.hpp" @@ -1062,19 +1063,17 @@ 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(fd_set_adapter& fd_set); + YASIO__DECL void process_channels(fd_set_adapter& fd_set); 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 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*, fd_set_adapter& fd_set); #if defined(YASIO_SSL_BACKEND) YASIO__DECL void init_ssl_context(); @@ -1086,7 +1085,8 @@ 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 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(); YASIO__DECL void destroy_ares_channel(); @@ -1100,13 +1100,13 @@ 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 unregister_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); - YASIO__DECL bool do_read(transport_handle_t, fd_set* fds_array); + 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); @@ -1129,7 +1129,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*, fd_set_adapter& fd_set); YASIO__DECL static const char* strerror(int error); @@ -1172,16 +1172,7 @@ 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, - write_op, - except_op, - max_ops, - }; - fd_set fds_array_[max_ops]; + fd_set_adapter fd_set_; // options struct __unnamed_options {