////////////////////////////////////////////////////////////////////////////////////////// // A multi-platform support c++11 library with focus on asynchronous socket I/O for any // client application. ////////////////////////////////////////////////////////////////////////////////////////// /* The MIT License (MIT) Copyright (c) 2012-2023 HALX99 HAL: Hardware Abstraction Layer X99: Intel X99 Mainboard Platform Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ #ifndef YASIO__IO_SERVICE_CPP #define YASIO__IO_SERVICE_CPP #if !defined(YASIO_HEADER_ONLY) # include "yasio/io_service.hpp" #endif #include #include #include #include #include #include "yasio/thread_name.hpp" #if defined(YASIO_SSL_BACKEND) # include "yasio/ssl.hpp" #endif #if defined(YASIO_ENABLE_KCP) # include "kcp/ikcp.h" struct yasio_kcp_options { int kcp_conv_ = 0; int kcp_nodelay_ = 1; int kcp_interval_ = 10; // 10~100ms int kcp_resend_ = 2; int kcp_ncwnd_ = 1; int kcp_sndwnd_ = 32; int kcp_rcvwnd_ = 128; int kcp_mtu_ = 1400; // kcp fast model the RTO min is 30. int kcp_minrto_ = 30; }; #endif #if defined(YASIO_USE_CARES) # include "yasio/impl/ares.hpp" #endif // clang-format off #define YASIO_KLOG_CP(level, format, ...) \ do \ { \ auto& __cprint = __get_cprint(); \ auto __msg = ::yasio::strfmt(127, "[yasio][%lld]" format "\n", ::yasio::clock(), ##__VA_ARGS__); \ if (__cprint) \ __cprint(level, __msg.c_str()); \ else \ YASIO_LOG_TAG("", "%s", __msg.c_str()); \ } while (false) // clang-format on #define YASIO_KLOGD(format, ...) YASIO_KLOG_CP(YLOG_D, format, ##__VA_ARGS__) #define YASIO_KLOGI(format, ...) YASIO_KLOG_CP(YLOG_I, format, ##__VA_ARGS__) #define YASIO_KLOGW(format, ...) YASIO_KLOG_CP(YLOG_W, format, ##__VA_ARGS__) #define YASIO_KLOGE(format, ...) YASIO_KLOG_CP(YLOG_E, format, ##__VA_ARGS__) #if !defined(YASIO_VERBOSE_LOG) # define YASIO_KLOGV(fmt, ...) (void)0 #else # define YASIO_KLOGV(format, ...) YASIO_KLOG_CP(YLOG_V, format, ##__VA_ARGS__) #endif #if defined(_MSC_VER) # pragma warning(push) # pragma warning(disable : 6320 6322 4996) #endif namespace yasio { YASIO__NS_INLINE namespace inet { namespace { enum : uint8_t { // op masks and stop flags YOPM_OPEN = 1, YOPM_CLOSE = 1 << 1, YSTF_STOP = 1 << 2, YSTF_FINALIZE = 1 << 3 }; enum { /* whether udp server enable multicast service */ YCPF_MCAST = 1 << 17, /* whether multicast loopback, if 1, local machine can recv self multicast packet */ YCPF_MCAST_LOOPBACK = 1 << 18, /* whether host dirty */ YCPF_HOST_DIRTY = 1 << 19, /* whether port dirty */ YCPF_PORT_DIRTY = 1 << 20, /* host is domain name, needs query via internet */ YCPF_NEEDS_QUERY = 1 << 21, /// below is byte2 of private flags (25~32) are mutable, and will be cleared automatically when connect flow done, see clear_mutable_flags. }; namespace { // By default we will wait no longer than 5 minutes. This will ensure that // any changes to the system clock are detected after no longer than this. static const highp_time_t yasio__max_wait_usec = 5 * 60 * 1000 * 1000LL; // the max transport alloc size static const size_t yasio__max_tsize = (std::max)({sizeof(io_transport_tcp), sizeof(io_transport_udp), sizeof(io_transport_ssl), sizeof(io_transport_kcp)}); } // namespace struct yasio__global_state { enum { INITF_SSL = 1, INITF_CARES = 2, }; yasio__global_state(const print_fn2_t& custom_print) { auto __get_cprint = [&]() -> const print_fn2_t& { return custom_print; }; #if defined(YASIO_SSL_BACKEND) && YASIO_SSL_BACKEND == 1 # if OPENSSL_VERSION_NUMBER >= 0x10100000 && !defined(LIBRESSL_VERSION_NUMBER) if (OPENSSL_init_ssl(0, nullptr) == 1) { yasio__setbits(this->init_flags_, INITF_SSL); ERR_clear_error(); } # endif #endif #if defined(YASIO_USE_CARES) int ares_status = ::ares_library_init(ARES_LIB_INIT_ALL); if (ares_status == 0) yasio__setbits(init_flags_, INITF_CARES); else YASIO_KLOGI("[global] c-ares library init failed, status=%d, detail:%s", ares_status, ::ares_strerror(ares_status)); # if defined(__ANDROID__) ares_status = ::yasio__ares_init_android(); if (ares_status != 0) YASIO_KLOGI("[global] c-ares library init android failed, status=%d, detail:%s", ares_status, ::ares_strerror(ares_status)); # endif #endif // print version & transport alloc size YASIO_KLOGI("[global] the yasio-%x.%x.%x is initialized, the size of per transport is %d when object_pool enabled.", (YASIO_VERSION_NUM >> 16) & 0xff, (YASIO_VERSION_NUM >> 8) & 0xff, YASIO_VERSION_NUM & 0xff, yasio__max_tsize); YASIO_KLOGI("[global] sizeof(io_event)=%d", sizeof(io_event)); } ~yasio__global_state() { #if defined(YASIO_USE_CARES) if (yasio__testbits(this->init_flags_, INITF_CARES)) ::ares_library_cleanup(); #endif } int init_flags_ = 0; print_fn2_t cprint_; }; static yasio__global_state& yasio__shared_globals(const print_fn2_t& prt = nullptr) { static yasio__global_state __global_state(prt); return __global_state; } } // namespace /// highp_timer void highp_timer::async_wait(timer_cb_t cb) { service_.schedule_timer(this, std::move(cb)); } void highp_timer::cancel() { if (!expired()) service_.remove_timer(this); } std::chrono::microseconds highp_timer::wait_duration() const { return std::chrono::duration_cast(this->expire_time_ - service_.time_); } /// io_send_op int io_send_op::perform(io_transport* transport, const void* buf, int n, int& error) { return transport->write_cb_(buf, n, nullptr, error); } /// io_sendto_op int io_sendto_op::perform(io_transport* transport, const void* buf, int n, int& error) { return transport->write_cb_(buf, n, std::addressof(destination_), error); } /// io_channel io_channel::io_channel(io_service& service, int index) : io_base(), service_(service), timer_(service), user_timer_(service) { socket_ = std::make_shared(); state_ = io_base::state::CLOSED; index_ = index; decode_len_ = [this](void* ptr, int len) { return this->__builtin_decode_len(ptr, len); }; } #if defined(YASIO_ENABLE_KCP) io_channel::~io_channel() { if (kcp_options_) delete kcp_options_; } yasio_kcp_options& io_channel::kcp_options() { if (!kcp_options_) kcp_options_ = new yasio_kcp_options(); return *kcp_options_; } #endif #if defined(YASIO_SSL_BACKEND) yssl_ctx_st* io_channel::get_ssl_context(bool client) const { if (client) return service_.ssl_roles_[YSSL_CLIENT]; auto& ctx = service_.ssl_roles_[YSSL_SERVER]; return (ctx) ? ctx : service_.init_ssl_context(YSSL_SERVER); } #endif const print_fn2_t& io_channel::__get_cprint() const { return get_service().options_.print_; } std::string io_channel::format_destination() const { if (yasio__testbits(properties_, YCPF_NEEDS_QUERY)) return yasio::strfmt(127, "%s(%s):%u", remote_host_.c_str(), !remote_eps_.empty() ? remote_eps_[0].ip().c_str() : "undefined", remote_port_); return yasio::strfmt(127, "%s:%u", remote_host_.c_str(), remote_port_); } void io_channel::enable_multicast(const char* addr, int loopback) { yasio__setbits(properties_, YCPF_MCAST); if (loopback) yasio__setbits(properties_, YCPF_MCAST_LOOPBACK); if (addr) multiaddr_.as_in(addr, (u_short)0); } void io_channel::join_multicast_group() { if (socket_->is_open()) { // interface switch (multiif_.af()) { case AF_INET: socket_->set_optval(IPPROTO_IP, IP_MULTICAST_IF, multiif_.in4_.sin_addr); break; case AF_INET6: socket_->set_optval(IPPROTO_IPV6, IP_MULTICAST_IF, multiif_.in6_.sin6_scope_id); break; } int loopback = yasio__testbits(properties_, YCPF_MCAST_LOOPBACK) ? 1 : 0; socket_->set_optval(multiaddr_.af() == AF_INET ? IPPROTO_IP : IPPROTO_IPV6, multiaddr_.af() == AF_INET ? IP_MULTICAST_LOOP : IPV6_MULTICAST_LOOP, loopback); // ttl socket_->set_optval(multiaddr_.af() == AF_INET ? IPPROTO_IP : IPPROTO_IPV6, multiaddr_.af() == AF_INET ? IP_MULTICAST_TTL : IPV6_MULTICAST_HOPS, YASIO_DEFAULT_MULTICAST_TTL); if (configure_multicast_group(true) != 0) { int ec = xxsocket::get_last_errno(); YASIO_KLOGE("[index: %d] join to multicast group %s failed, ec=%d, detail:%s", this->index_, multiaddr_.to_string().c_str(), ec, xxsocket::strerror(ec)); } } } void io_channel::disable_multicast() { yasio__clearbits(properties_, YCPF_MCAST); yasio__clearbits(properties_, YCPF_MCAST_LOOPBACK); if (socket_->is_open()) configure_multicast_group(false); } int io_channel::configure_multicast_group(bool onoff) { if (multiaddr_.af() == AF_INET) { // ipv4 struct ip_mreq mreq; mreq.imr_interface.s_addr = multiif_.in4_.sin_addr.s_addr; mreq.imr_multiaddr = multiaddr_.in4_.sin_addr; return socket_->set_optval(IPPROTO_IP, onoff ? IP_ADD_MEMBERSHIP : IP_DROP_MEMBERSHIP, &mreq, (int)sizeof(mreq)); } else { // ipv6 struct ipv6_mreq mreq_v6; mreq_v6.ipv6mr_interface = multiif_.in6_.sin6_scope_id; mreq_v6.ipv6mr_multiaddr = multiaddr_.in6_.sin6_addr; return socket_->set_optval(IPPROTO_IPV6, onoff ? IPV6_JOIN_GROUP : IPV6_LEAVE_GROUP, &mreq_v6, (int)sizeof(mreq_v6)); } } void io_channel::set_host(cxx17::string_view host) { if (this->remote_host_ != host) { cxx17::assign(this->remote_host_, host); yasio__setbits(properties_, YCPF_HOST_DIRTY); } } void io_channel::set_port(u_short port) { if (port == 0) return; if (this->remote_port_ != port) { this->remote_port_ = port; yasio__setbits(properties_, YCPF_PORT_DIRTY); } } int io_channel::__builtin_decode_len(void* d, int n) { int loffset = uparams_.length_field_offset; int lsize = uparams_.length_field_length; if (loffset >= 0) { assert(lsize >= 1 && lsize <= YASIO_SSIZEOF(int)); int len = 0; if (n >= (loffset + lsize)) { ::memcpy(&len, (uint8_t*)d + loffset, lsize); if (!uparams_.no_bswap) len = yasio::network_to_host(len, lsize); len += uparams_.length_adjustment; if (len > uparams_.max_frame_length) len = -1; } return len; } return n; } // -------------------- io_transport --------------------- io_transport::io_transport(io_channel* ctx, xxsocket_ptr&& s) : ctx_(ctx) { this->state_ = io_base::state::OPENED; this->socket_ = std::move(s); #if !defined(YASIO_MINIFY_EVENT) this->ud_.ptr = nullptr; #endif } const print_fn2_t& io_transport::__get_cprint() const { return ctx_->get_service().options_.print_; } int io_transport::write(io_send_buffer&& buffer, completion_cb_t&& handler) { int n = static_cast(buffer.size()); send_queue_.emplace(cxx14::make_unique(std::move(buffer), std::move(handler))); get_service().wakeup(); return n; } int io_transport::do_read(int revent, int& error, highp_time_t&) { return this->call_read(buffer_ + offset_, sizeof(buffer_) - offset_, revent, error); } bool io_transport::do_write(highp_time_t& wait_duration) { bool ret = false; do { if (!socket_->is_open()) break; int error = 0; auto wrap = send_queue_.peek(); if (wrap) { auto& v = *wrap; if (call_write(v.get(), error) < 0) { this->set_last_errno(error, yasio::io_base::error_stage::WRITE); break; } } bool no_wevent = send_queue_.empty(); if (yasio__unlikely(!no_wevent)) { // still have work to do no_wevent = (error != EWOULDBLOCK && error != EAGAIN && error != ENOBUFS); if (!no_wevent) { // system kernel buffer full if (!pollout_registerred_) { get_service().io_watcher_.mod_event(socket_->native_handle(), socket_event::write, 0); pollout_registerred_ = true; } } else wait_duration = 0; } if (no_wevent && pollout_registerred_) { get_service().io_watcher_.mod_event(socket_->native_handle(), 0, socket_event::write); pollout_registerred_ = false; } ret = true; } while (false); return ret; } int io_transport::call_read(void* data, int size, int revent, int& error) { int n = read_cb_(data, size, revent, error); if (n > 0) { ctx_->bytes_transferred_ += n; return n; } if (n < 0) { if (xxsocket::not_recv_error(error)) return (error = 0); // status ok, clear error return n; } if (yasio__testbits(ctx_->properties_, YCM_TCP)) { error = yasio::errc::eof; return -1; } return 0; } int io_transport::call_write(io_send_op* op, int& error) { int n = op->perform(this, op->buffer_.data() + op->offset_, static_cast(op->buffer_.size() - op->offset_), error); if (n > 0) { // #performance: change offset only, remain data will be send at next frame. op->offset_ += n; if (op->offset_ == op->buffer_.size()) this->complete_op(op, 0); } else if (n < 0) { if (xxsocket::not_send_error(error)) n = 0; else if (yasio__testbits(ctx_->properties_, YCM_UDP)) { // !!! For udp, simply drop the op instead trigger handle close, // on android device, the error will be 'EPERM' when app in background. this->complete_op(op, error); n = 0; } } return n; } void io_transport::complete_op(io_send_op* op, int error) { YASIO_KLOGV("[index: %d] write complete, bytes transferred: %d/%d", this->cindex(), static_cast(op->offset_), static_cast(op->buffer_.size())); if (op->handler_) op->handler_(error, op->offset_); send_queue_.pop(); } void io_transport::set_primitives() { this->write_cb_ = [this](const void* data, int len, const ip::endpoint*, int& error) { int n = socket_->send(data, len, YASIO_MSG_FLAG); if (n < 0) error = xxsocket::get_last_errno(); return n; }; this->read_cb_ = [this](void* data, int len, int revent, int& error) { if (revent) { int n = socket_->recv(data, len); if (n < 0) error = xxsocket::get_last_errno(); return n; } error = EWOULDBLOCK; return -1; }; } // -------------------- io_transport_tcp --------------------- inline io_transport_tcp::io_transport_tcp(io_channel* ctx, xxsocket_ptr&& s) : io_transport(ctx, std::forward(s)) {} // ----------------------- io_transport_ssl ---------------- #if defined(YASIO_SSL_BACKEND) io_transport_ssl::io_transport_ssl(io_channel* ctx, xxsocket_ptr&& sock) : io_transport_tcp(ctx, std::forward(sock)) { this->state_ = io_base::state::CONNECTING; // for ssl, inital state shoud be connecing for ssl handshake bool client = yasio__testbits(ctx->properties_, YCM_CLIENT); this->ssl_ = yssl_new(ctx->get_ssl_context(client), static_cast(this->socket_->native_handle()), ctx->remote_host_.c_str(), client); } int io_transport_ssl::do_ssl_handshake(int& error) { int ret = yssl_do_handshake(ssl_, error); if (ret == 0) // handshake succeed { // because we invoke handshake in call_read, so we emit EWOULDBLOCK to mark ssl transport status `ok` this->state_ = io_base::state::OPENED; this->read_cb_ = [this](void* data, int len, int revent, int& error) { if (revent) return yssl_read(ssl_, data, len, error); error = EWOULDBLOCK; return -1; }; this->write_cb_ = [this](const void* data, int len, const ip::endpoint*, int& error) { return yssl_write(ssl_, data, len, error); }; YASIO_KLOGD("[index: %d] the connection #%u <%s> --> <%s> is established.", ctx_->index_, this->id_, this->local_endpoint().to_string().c_str(), this->remote_endpoint().to_string().c_str()); get_service().fire_event(ctx_->index_, YEK_ON_OPEN, 0, this); error = EWOULDBLOCK; } else { if (error == EWOULDBLOCK) get_service().wakeup(); else { // handshake failed, print reason char buf[256] = {0}; YASIO_KLOGE("[index: %d] do_ssl_handshake fail with %s", ctx_->index_, yssl_strerror(ssl_, ret, buf, sizeof(buf))); if (yasio__testbits(ctx_->properties_, YCM_CLIENT)) { YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx_->index_, ctx_->format_destination().c_str(), error, io_service::strerror(error)); get_service().fire_event(ctx_->index(), YEK_ON_OPEN, error, ctx_); } } } return -1; } void io_transport_ssl::do_ssl_shutdown() { if (ssl_) yssl_shutdown(ssl_, this->error_ == yasio::errc::shutdown_by_localhost); } void io_transport_ssl::set_primitives() { this->read_cb_ = [this](void* /*data*/, int /*len*/, int /*revent*/, int& error) { return do_ssl_handshake(error); }; } #endif // ----------------------- io_transport_udp ---------------- io_transport_udp::io_transport_udp(io_channel* ctx, xxsocket_ptr&& s) : io_transport(ctx, std::forward(s)) {} io_transport_udp::~io_transport_udp() {} ip::endpoint io_transport_udp::remote_endpoint() const { return !connected_ ? this->peer_ : socket_->peer_endpoint(); } const ip::endpoint& io_transport_udp::ensure_destination() const { if (this->destination_.af() != AF_UNSPEC) return this->destination_; return (this->destination_ = this->peer_); } void io_transport_udp::confgure_remote(const ip::endpoint& peer) { if (connected_) // connected, update peer is pointless and useless return; this->peer_ = peer; if (!yasio__testbits(ctx_->properties_, YCPF_MCAST) || !yasio__testbits(ctx_->properties_, YCM_CLIENT)) this->connect(); // multicast client, don't bind multicast address for we can recvfrom non-multicast address } void io_transport_udp::connect() { if (connected_) return; if (this->peer_.af() == AF_UNSPEC) { if (ctx_->remote_eps_.empty()) return; this->peer_ = ctx_->remote_eps_[0]; } int retval = this->socket_->connect_n(this->peer_); connected_ = (retval == 0); set_primitives(); } void io_transport_udp::disconnect() { #if defined(__linux__) auto ifaddr = this->socket_->local_endpoint(); #endif int retval = this->socket_->disconnect(); #if defined(__linux__) if (retval == 0) { // Because some of linux will unbind when disconnect succeed, so try to rebind ifaddr.ip(ctx_->local_host_.empty() ? YASIO_ADDR_ANY(ifaddr.af()) : ctx_->local_host_.c_str()); this->socket_->bind(ifaddr); } #else YASIO__UNUSED_PARAM(retval); #endif connected_ = false; set_primitives(); } int io_transport_udp::write(io_send_buffer&& buffer, completion_cb_t&& handler) { return connected_ ? io_transport::write(std::move(buffer), std::move(handler)) : write_to(std::move(buffer), ensure_destination(), std::move(handler)); } int io_transport_udp::write_to(io_send_buffer&& buffer, const ip::endpoint& to, completion_cb_t&& handler) { int n = static_cast(buffer.size()); send_queue_.emplace(cxx14::make_unique(std::move(buffer), std::move(handler), to)); get_service().wakeup(); return n; } void io_transport_udp::set_primitives() { if (connected_) io_transport::set_primitives(); else { this->write_cb_ = [this](const void* data, int len, const ip::endpoint* destination, int& error) { assert(destination); int n = socket_->sendto(data, len, *destination); if (n < 0) { error = xxsocket::get_last_errno(); if (!xxsocket::not_send_error(error)) YASIO_KLOGW("[index: %d] write udp socket failed, ec=%d, detail:%s", this->cindex(), error, io_service::strerror(error)); } return n; }; this->read_cb_ = [this](void* data, int len, int revent, int& error) { if (revent) { ip::endpoint peer; int n = socket_->recvfrom(data, len, peer); if (n > 0) this->peer_ = peer; if (n < 0) error = xxsocket::get_last_errno(); return n; } error = EWOULDBLOCK; return -1; }; } } int io_transport_udp::handle_input(const char* data, int bytes_transferred, int& /*error*/, highp_time_t&) { // pure udp, dispatch to upper layer directly get_service().fire_event(this->cindex(), io_packet{data, data + bytes_transferred}, this); return bytes_transferred; } #if defined(YASIO_ENABLE_KCP) // ----------------------- io_transport_kcp ------------------ io_transport_kcp::io_transport_kcp(io_channel* ctx, xxsocket_ptr&& s) : io_transport_udp(ctx, std::forward(s)) { auto& kopts = ctx->kcp_options(); this->kcp_ = ::ikcp_create(static_cast(kopts.kcp_conv_), this); ::ikcp_nodelay(this->kcp_, kopts.kcp_nodelay_, kopts.kcp_interval_ /*kcp max interval is 5000(ms)*/, kopts.kcp_resend_, kopts.kcp_ncwnd_); ::ikcp_wndsize(this->kcp_, kopts.kcp_sndwnd_, kopts.kcp_rcvwnd_); ::ikcp_setmtu(this->kcp_, kopts.kcp_mtu_); // Because of nodelaying config will change the value. so setting RTO min after call ikcp_nodely. this->kcp_->rx_minrto = kopts.kcp_minrto_; this->rawbuf_.resize(YASIO_INET_BUFFER_SIZE); ::ikcp_setoutput(this->kcp_, [](const char* buf, int len, ::ikcpcb* /*kcp*/, void* user) { auto t = (io_transport_kcp*)user; int ignored_ec = 0; return t->write_cb_(buf, len, std::addressof(t->ensure_destination()), ignored_ec); }); } io_transport_kcp::~io_transport_kcp() { ::ikcp_release(this->kcp_); } int io_transport_kcp::write(io_send_buffer&& buffer, completion_cb_t&& handler) { std::lock_guard lck(send_mtx_); int nsent = ::ikcp_send(kcp_, buffer.data(), static_cast(buffer.size())); assert(nsent > 0); if (handler) handler(nsent > 0 ? 0 : nsent, nsent); get_service().wakeup(); return nsent; } int io_transport_kcp::do_read(int revent, int& error, highp_time_t& wait_duration) { int n = this->call_read(&rawbuf_.front(), static_cast(rawbuf_.size()), revent, error); if (n > 0) this->handle_input(rawbuf_.data(), n, error, wait_duration); if (!error) { // !important, should always try to call ikcp_recv when no error occured. n = ::ikcp_recv(kcp_, buffer_ + offset_, sizeof(buffer_) - offset_); if (n > 0) // If got data from kcp, don't wait wait_duration = 0; else if (n < 0) n = 0; // EAGAIN/EWOULDBLOCK } return n; } int io_transport_kcp::handle_input(const char* buf, int len, int& error, highp_time_t& wait_duration) { // ikcp in event always in service thread, so no need to lock if (0 == ::ikcp_input(kcp_, buf, len)) { this->check_timeout(wait_duration); // call ikcp_check return len; } // simply regards -1,-2,-3 as error and trigger connection lost event. error = yasio::errc::invalid_packet; return -1; } bool io_transport_kcp::do_write(highp_time_t& wait_duration) { std::lock_guard lck(send_mtx_); ::ikcp_update(kcp_, static_cast(::yasio::clock())); ::ikcp_flush(kcp_); this->check_timeout(wait_duration); // call ikcp_check return true; } static IINT32 yasio_itimediff(IUINT32 later, IUINT32 earlier) { return static_cast(later - earlier); } static IUINT32 yasio_ikcp_check(const ikcpcb* kcp, IUINT32 current, IUINT32 waitd_ms) { IUINT32 ts_flush = kcp->ts_flush; IINT32 tm_flush = 0x7fffffff; IINT32 tm_packet = 0x7fffffff; IUINT32 minimal = 0; struct IQUEUEHEAD* p; if (kcp->updated == 0) return current; if (yasio_itimediff(current, ts_flush) < -10000) ts_flush = current; if (yasio_itimediff(current, ts_flush) >= 0) return current; if (kcp->nsnd_que) return current; if (kcp->probe) return current; if (kcp->rmt_wnd == 0 && yasio_itimediff(kcp->current, kcp->ts_probe) >= 0) return current; tm_flush = yasio_itimediff(ts_flush, current); for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = p->next) { const IKCPSEG* seg = iqueue_entry(p, const IKCPSEG, node); IINT32 diff = yasio_itimediff(seg->resendts, current); if (diff <= 0) { return current; } if (diff < tm_packet) tm_packet = diff; } minimal = kcp->nsnd_buf ? static_cast(tm_packet < tm_flush ? tm_packet : tm_flush) : waitd_ms; return current + minimal; } void io_transport_kcp::check_timeout(highp_time_t& wait_duration) const { auto current = static_cast(::yasio::clock()); auto expire_time = yasio_ikcp_check(kcp_, current, static_cast(wait_duration / std::milli::den)); highp_time_t duration = static_cast(expire_time - current) * std::milli::den; if (duration < 0) duration = 0; if (wait_duration > duration) wait_duration = duration; } #endif // ------------------------ io_service ------------------------ void io_service::init_globals(const yasio::inet::print_fn2_t& prt) { yasio__shared_globals(prt).cprint_ = prt; } void io_service::cleanup_globals() { yasio__shared_globals().cprint_ = nullptr; } unsigned int io_service::tcp_rtt(transport_handle_t transport) { return transport->is_open() ? transport->socket_->tcp_rtt() : 0; } io_service::io_service() { this->initialize(nullptr, 1); } io_service::io_service(int channel_count) { this->initialize(nullptr, channel_count); } io_service::io_service(const io_hostent& channel_ep) { this->initialize(&channel_ep, 1); } io_service::io_service(const std::vector& channel_eps) { this->initialize(!channel_eps.empty() ? channel_eps.data() : nullptr, static_cast(channel_eps.size())); } io_service::io_service(const io_hostent* channel_eps, int channel_count) { this->initialize(channel_eps, channel_count); } io_service::~io_service() { this->do_stop(YSTF_FINALIZE); this->finalize(); } void io_service::start(event_cb_t cb) { if (state_ == io_service::state::IDLE) { auto& global_state = yasio__shared_globals(); if (!this->options_.print_) this->options_.print_ = global_state.cprint_; if (cb) options_.on_event_ = std::move(cb); this->state_ = io_service::state::RUNNING; if (!options_.no_new_thread_) { this->worker_ = std::thread(&io_service::run, this); this->worker_id_ = worker_.get_id(); } else { this->worker_id_ = std::this_thread::get_id(); run(); handle_stop(); } } } void io_service::stop() { do_stop(YSTF_STOP); } void io_service::do_stop(uint8_t flags) { if (this->state_ <= io_service::state::IDLE) return; if (!this->stop_flag_) { this->stop_flag_ = flags; if (this->state_ == io_service::state::RUNNING) { for (auto c : channels_) this->close(c->index()); this->wakeup(); this->handle_stop(); } } else this->handle_stop(); } void io_service::handle_stop() { if (this->worker_.joinable()) { if (std::this_thread::get_id() == this->worker_id_) { if (yasio__testbits(this->stop_flag_, YSTF_FINALIZE)) std::terminate(); // we don't want, but... xxsocket::set_last_errno(EAGAIN); return; } this->worker_.join(); } if (this->state_ != state::AT_EXITING) return; if (this->options_.deferred_event_ && !this->events_.empty()) this->dispatch((std::numeric_limits::max)()); clear_transports(); this->timer_queue_.clear(); this->stop_flag_ = 0; this->worker_id_ = std::thread::id{}; this->state_ = io_service::state::IDLE; } void io_service::initialize(const io_hostent* channel_eps, int channel_count) { #if defined(YASIO_SSL_BACKEND) ssl_roles_[YSSL_CLIENT] = ssl_roles_[YSSL_SERVER] = nullptr; #endif this->wait_duration_ = yasio__max_wait_usec; // at least one channel if (channel_count < 1) channel_count = 1; options_.resolv_ = [this](std::vector& eps, const char* host, unsigned short port) { return this->resolve(eps, host, port); }; // create channels create_channels(channel_eps, channel_count); #if !defined(YASIO_USE_CARES) life_mutex_ = std::make_shared(); life_token_ = std::make_shared(); #endif this->state_ = io_service::state::IDLE; } void io_service::finalize() { if (this->state_ == io_service::state::IDLE) { #if !defined(YASIO_USE_CARES) std::unique_lock lck(*life_mutex_); life_token_.reset(); #endif destroy_channels(); options_.on_event_ = nullptr; options_.resolv_ = nullptr; for (auto o : tpool_) ::operator delete(o); tpool_.clear(); this->state_ = io_service::state::UNINITIALIZED; } } void io_service::create_channels(const io_hostent* channel_eps, int channel_count) { for (auto i = 0; i < channel_count; ++i) { auto channel = new io_channel(*this, i); if (channel_eps != nullptr) channel->set_address(channel_eps[i].host_, channel_eps[i].port_); channels_.push_back(channel); } } void io_service::destroy_channels() { this->channel_ops_.clear(); for (auto channel : channels_) { channel->timer_.cancel(); cleanup_io(channel); delete channel; } channels_.clear(); } void io_service::clear_transports() { for (auto transport : transports_) { cleanup_io(transport); yasio::invoke_dtor(transport); this->tpool_.push_back(transport); } transports_.clear(); } size_t io_service::dispatch(int max_count) { if (options_.on_event_) this->events_.consume(max_count, options_.on_event_); return this->events_.count(); } void io_service::run() { yasio::set_thread_name("yasio"); #if defined(YASIO_SSL_BACKEND) init_ssl_context(YSSL_CLIENT); // by default, init ssl client context #endif #if defined(YASIO_USE_CARES) recreate_ares_channel(); ares_socket_t ares_socks[ARES_GETSOCK_MAXNUM] = {0}; #endif // Call once at startup this->ipsv_ = static_cast(xxsocket::getipsv()); // Update time for 1st loop this->update_time(); do { auto waitd_usec = get_timeout(this->wait_duration_); // Gets current wait duration #if defined(YASIO_USE_CARES) /** * retrieves the set of file descriptors which the calling application should poll io, * after poll_io, for ares invoke flow, refer to: * https://c-ares.org/ares_fds.html * https://c-ares.org/ares_timeout.html * https://c-ares.org/ares_process_fd.html */ auto ares_nfds = ares_get_fds(ares_socks, waitd_usec); #endif if (waitd_usec > 0) { YASIO_KLOGV("[core] poll_io max_nfds=%d, waiting... %.3f milliseconds", io_watcher_.max_descriptor(), waitd_usec / static_cast(std::milli::den)); int retval = io_watcher_.poll_io(waitd_usec); YASIO_KLOGV("[core] poll_io waked up, retval=%d", retval); if (retval < 0) { int ec = xxsocket::get_last_errno(); YASIO_KLOGI("[core] poll_io failed, max_fd=%d ec=%d, detail:%s\n", io_watcher_.max_descriptor(), ec, io_service::strerror(ec)); if (ec != EBADF) continue; // Try again. break; } } #if defined(YASIO_USE_CARES) // process events for name resolution. do_ares_process_fds(ares_socks, ares_nfds); #endif // process active transports process_transports(); // process active channels process_channels(); // process timeout timers process_timers(); // process deferred events if auto dispatch enabled process_deferred_events(); } while (!this->stop_flag_ || !this->transports_.empty()); #if defined(YASIO_USE_CARES) destroy_ares_channel(); #endif #if defined(YASIO_SSL_BACKEND) cleanup_ssl_context(YSSL_CLIENT); cleanup_ssl_context(YSSL_SERVER); #endif this->state_ = io_service::state::AT_EXITING; } void io_service::process_transports() { // preform transports for (auto iter = transports_.begin(); iter != transports_.end();) { auto transport = *iter; bool ok = (do_read(transport) && do_write(transport)); if (ok) { int opm = transport->opmask_ | transport->ctx_->opmask_ | this->stop_flag_; if (0 == opm) { // no open/close/stop operations request ++iter; continue; } if (transport->error_ == 0) transport->error_ = yasio::errc::shutdown_by_localhost; } handle_close(transport); iter = transports_.erase(iter); } } void io_service::process_channels() { if (!this->channel_ops_.empty()) { // perform active channels std::lock_guard lck(this->channel_ops_mtx_); for (auto iter = this->channel_ops_.begin(); iter != this->channel_ops_.end();) { auto ctx = *iter; bool finish = true; if (yasio__testbits(ctx->properties_, YCM_CLIENT)) { if (yasio__testbits(ctx->opmask_, YOPM_OPEN)) { yasio__clearbits(ctx->opmask_, YOPM_OPEN); ctx->state_ = io_base::state::RESOLVING; } if (ctx->state_ == io_base::state::RESOLVING) { if (do_resolve(ctx) == 0) do_connect(ctx); else if (ctx->error_ != EINPROGRESS) handle_connect_failed(ctx, ctx->error_); } else if (ctx->state_ == io_base::state::CONNECTING) do_connect_completion(ctx); finish = ctx->error_ != EINPROGRESS; } else if (yasio__testbits(ctx->properties_, YCM_SERVER)) { auto opmask = ctx->opmask_; if (yasio__testbits(opmask, YOPM_OPEN)) do_accept(ctx); else if (yasio__testbits(opmask, YOPM_CLOSE)) cleanup_channel(ctx); finish = (ctx->state_ != io_base::state::OPENED); if (!finish) do_accept_completion(ctx); else ctx->bytes_transferred_ = 0; } if (finish) iter = this->channel_ops_.erase(iter); else ++iter; } } } void io_service::close(int index) { // Gets channel context auto channel = channel_at(index); if (!channel) return; if (!yasio__testbits(channel->opmask_, YOPM_CLOSE)) { yasio__clearbits(channel->opmask_, YOPM_OPEN); if (channel->socket_->is_open()) { yasio__setbits(channel->opmask_, YOPM_CLOSE); this->wakeup(); } } } void io_service::close(transport_handle_t transport) { if (!yasio__testbits(transport->opmask_, YOPM_CLOSE)) { yasio__setbits(transport->opmask_, YOPM_CLOSE); this->wakeup(); } } bool io_service::is_open(transport_handle_t transport) const { return transport->is_open(); } bool io_service::is_open(int index) const { auto ctx = channel_at(index); return ctx != nullptr && ctx->state_ == io_base::state::OPENED; } bool io_service::open(size_t index, int kind) { assert((kind > 0 && kind <= 0xff) && ((kind & (kind - 1)) != 0)); auto ctx = channel_at(index); if (ctx != nullptr) { yasio__setlobyte(ctx->properties_, kind & 0xff); if (yasio__testbits(kind, YCM_TCP)) ctx->socktype_ = SOCK_STREAM; else if (yasio__testbits(kind, YCM_UDP)) ctx->socktype_ = SOCK_DGRAM; return open_internal(ctx); } return false; } io_channel* io_service::channel_at(size_t index) const { return (index < channels_.size()) ? channels_[index] : nullptr; } void io_service::handle_close(transport_handle_t thandle) { auto ctx = thandle->ctx_; auto error = thandle->error_; const bool client = yasio__testbits(ctx->properties_, YCM_CLIENT); if (thandle->state_ == io_base::state::OPENED) { // @Because we can't retrive peer endpoint when connect reset by peer, so use id to trace. YASIO_KLOGD("[index: %d] the connection #%u is lost, ec=%d, where=%d, detail:%s", ctx->index_, thandle->id_, error, (int)thandle->error_stage_, io_service::strerror(error)); this->fire_event(ctx->index(), YEK_ON_CLOSE, error, thandle); } #if defined(YASIO_SSL_BACKEND) if (yasio__testbits(ctx->properties_, YCM_SSL)) static_cast(thandle)->do_ssl_shutdown(); #endif if (yasio__testbits(ctx->properties_, YCM_TCP) && error == yasio::errc::shutdown_by_localhost) thandle->socket_->shutdown(); cleanup_io(thandle); deallocate_transport(thandle); if (client) { yasio__clearbits(ctx->opmask_, YOPM_CLOSE); cleanup_channel(ctx, false); } } int io_service::write(transport_handle_t transport, sbyte_buffer buffer, completion_cb_t handler) { if (transport && transport->is_open()) return !buffer.empty() ? transport->write(io_send_buffer{std::move(buffer)}, std::move(handler)) : 0; else { YASIO_KLOGE("write failed, the connection not ok!"); return -1; } } int io_service::forward(transport_handle_t transport, const void* buf, size_t len, completion_cb_t handler) { if (transport && transport->is_open()) return len != 0 ? transport->write(io_send_buffer{(const char*)buf, len}, std::move(handler)) : 0; else { YASIO_KLOGE("write failed, the connection not ok!"); return -1; } } int io_service::write_to(transport_handle_t transport, sbyte_buffer buffer, const ip::endpoint& to, completion_cb_t handler) { if (transport && transport->is_open()) return !buffer.empty() ? transport->write_to(io_send_buffer{std::move(buffer)}, to, std::move(handler)) : 0; else { YASIO_KLOGE("write_to failed, the connection not ok!"); return -1; } } int io_service::forward_to(transport_handle_t transport, const void* buf, size_t len, const ip::endpoint& to, completion_cb_t handler) { if (transport && transport->is_open()) return len != 0 ? transport->write_to(io_send_buffer{(const char*)buf, len}, to, std::move(handler)) : 0; else { YASIO_KLOGE("write_to failed, the connection not ok!"); return -1; } } void io_service::do_connect(io_channel* ctx) { assert(!ctx->remote_eps_.empty()); if (this->ipsv_ == 0) this->ipsv_ = static_cast(xxsocket::getipsv()); if (ctx->socket_->is_open()) cleanup_io(ctx); ctx->state_ = io_base::state::CONNECTING; auto& ep = ctx->remote_eps_[0]; YASIO_KLOGD("[index: %d] connecting server %s(%s):%u...", ctx->index_, ctx->remote_host_.c_str(), ep.ip().c_str(), ctx->remote_port_); if (ctx->socket_->popen(ep.af(), ctx->socktype_)) { int ret = 0; if (yasio__testbits(ctx->properties_, YCF_REUSEADDR)) ctx->socket_->reuse_address(true); if (yasio__testbits(ctx->properties_, YCF_EXCLUSIVEADDRUSE)) ctx->socket_->exclusive_address(true); if (!yasio__testbits(ctx->properties_, YCM_UDS)) { auto ifaddr = ctx->local_host_.empty() ? YASIO_ADDR_ANY(ep.af()) : ctx->local_host_.c_str(); ret = ctx->socket_->bind(ifaddr, ctx->local_port_); } if (ret == 0) { // tcp connect directly, for udp do not need to connect. if (yasio__testbits(ctx->properties_, YCM_TCP)) ret = xxsocket::connect(ctx->socket_->native_handle(), ep); // join the multicast group for udp if (yasio__testbits(ctx->properties_, YCPF_MCAST)) ctx->join_multicast_group(); } if (ret < 0) { // setup non-blocking connect int error = xxsocket::get_last_errno(); if (error != EINPROGRESS && error != EWOULDBLOCK) this->handle_connect_failed(ctx, error); else { ctx->set_last_errno(EINPROGRESS); io_watcher_.mod_event(ctx->socket_->native_handle(), socket_event::readwrite, 0); ctx->timer_.expires_from_now(std::chrono::microseconds(options_.connect_timeout_)); ctx->timer_.async_wait_once([ctx](io_service& thiz) { if (ctx->state_ != io_base::state::OPENED) thiz.handle_connect_failed(ctx, ETIMEDOUT); }); } } else if (ret == 0) { // connect server successful immediately. io_watcher_.mod_event(ctx->socket_->native_handle(), socket_event::read, 0); handle_connect_succeed(ctx, ctx->socket_); } // !!!NEVER GO HERE } else this->handle_connect_failed(ctx, xxsocket::get_last_errno()); } void io_service::do_connect_completion(io_channel* ctx) { assert(ctx->state_ == io_base::state::CONNECTING); if (ctx->state_ == io_base::state::CONNECTING) { if (io_watcher_.is_ready(ctx->socket_->native_handle(), socket_event::readwrite)) { int error = -1; 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 io_watcher_.mod_event(ctx->socket_->native_handle(), 0, socket_event::write); handle_connect_succeed(ctx, ctx->socket_); } else handle_connect_failed(ctx, error); ctx->timer_.cancel(); } } } #if defined(YASIO_SSL_BACKEND) yssl_ctx_st* io_service::init_ssl_context(ssl_role role) { auto ctx = role == YSSL_CLIENT ? yssl_ctx_new(yssl_options{yasio__c_str(options_.cafile_), nullptr, true}) : yssl_ctx_new(yssl_options{yasio__c_str(options_.crtfile_), yasio__c_str(options_.keyfile_), false}); ssl_roles_[role] = ctx; return ctx; } void io_service::cleanup_ssl_context(ssl_role role) { auto& ctx = ssl_roles_[role]; if (ctx) yssl_ctx_free(ctx); } #endif #if defined(YASIO_USE_CARES) void io_service::ares_work_started() { ++ares_outstanding_work_; } void io_service::ares_work_finished() { if (ares_outstanding_work_ > 0) --ares_outstanding_work_; } void io_service::ares_sock_state_cb(void* data, socket_native_type socket_fd, int readable, int writable) { auto service = (io_service*)data; int events = socket_event::null; if (readable) events |= socket_event::read; if (writable) events |= socket_event::write; if (events != 0) service->io_watcher_.mod_event(socket_fd, events, 0); else service->io_watcher_.mod_event(socket_fd, 0, socket_event::readwrite); } void io_service::ares_getaddrinfo_cb(void* data, int status, int /*timeouts*/, ares_addrinfo* answerlist) { auto ctx = (io_channel*)data; auto& current_service = ctx->get_service(); current_service.ares_work_finished(); if (status == ARES_SUCCESS && answerlist != nullptr) { for (auto ai = answerlist->nodes; ai != nullptr; ai = ai->ai_next) { if (ai->ai_family == AF_INET6 || ai->ai_family == AF_INET) ctx->remote_eps_.push_back(ip::endpoint(ai->ai_addr)); } } auto __get_cprint = [&]() -> const print_fn2_t& { return current_service.options_.print_; }; if (!ctx->remote_eps_.empty()) { ctx->query_success_time_ = highp_clock(); # if defined(YASIO_ENABLE_ARES_PROFILER) YASIO_KLOGD("[index: %d] ares_getaddrinfo_cb: query %s succeed, cost:%g(ms)", ctx->index_, ctx->remote_host_.c_str(), (ctx->query_success_time_ - ctx->query_start_time_) / 1000.0); # endif } else { ctx->set_last_errno(yasio::errc::resolve_host_failed); YASIO_KLOGE("[index: %d] ares_getaddrinfo_cb: query %s failed, status=%d, detail:%s", ctx->index_, ctx->remote_host_.c_str(), status, ::ares_strerror(status)); } current_service.wakeup(); } int io_service::ares_get_fds(socket_native_type* socks, highp_time_t& waitd_usec) { int nfds = 0; if (ares_outstanding_work_) { int bitmask = ::ares_getsock(this->ares_, socks, ARES_GETSOCK_MAXNUM); for (int i = 0; i < ARES_GETSOCK_MAXNUM; ++i) { int events = socket_event::null; if (ARES_GETSOCK_READABLE(bitmask, i)) events |= socket_event::read; if (ARES_GETSOCK_WRITABLE(bitmask, i)) events |= socket_event::write; if (events) ++nfds; else break; } if (nfds) { timeval waitd_tv = {(decltype(timeval::tv_sec))(waitd_usec / std::micro::den), (decltype(timeval::tv_usec))(waitd_usec % std::micro::den)}; ::ares_timeout(this->ares_, &waitd_tv, &waitd_tv); waitd_usec = waitd_tv.tv_sec * std::micro::den + waitd_tv.tv_usec; } } return nfds; } void io_service::do_ares_process_fds(socket_native_type* socks, int nfds) { for (auto i = 0; i < nfds; ++i) { auto fd = socks[i]; ::ares_process_fd(this->ares_, io_watcher_.is_ready(fd, socket_event::read) ? fd : ARES_SOCKET_BAD, io_watcher_.is_ready(fd, socket_event::write) ? fd : ARES_SOCKET_BAD); } } void io_service::recreate_ares_channel() { if (ares_) destroy_ares_channel(); int optmask = ARES_OPT_TIMEOUTMS | ARES_OPT_TRIES | ARES_OPT_SOCK_STATE_CB /* | ARES_OPT_LOOKUPS*/; ares_options options = {}; options.timeout = static_cast(this->options_.dns_queries_timeout_ / std::milli::den); options.tries = this->options_.dns_queries_tries_; options.sock_state_cb = io_service::ares_sock_state_cb; options.sock_state_cb_data = this; # if defined(__linux__) && !defined(__ANDROID__) if (yasio::is_regular_file(YASIO_SYSTEMD_RESOLV_PATH)) { options.resolvconf_path = strndup(YASIO_SYSTEMD_RESOLV_PATH, YASIO_SYSTEMD_RESOLV_PATH_LEN); optmask |= ARES_OPT_RESOLVCONF; } # endif int status = ::ares_init_options(&ares_, &options, optmask); if (status == ARES_SUCCESS) { YASIO_KLOGD("[c-ares] create channel succeed"); config_ares_name_servers(); } else YASIO_KLOGE("[c-ares] create channel failed, status=%d, detail:%s", status, ::ares_strerror(status)); } void io_service::config_ares_name_servers() { std::string nscsv; // list all dns servers for resov problem diagnosis ares_addr_port_node* name_servers = nullptr; const char* what = "system"; if (!options_.name_servers_.empty()) { ::ares_set_servers_ports_csv(ares_, options_.name_servers_.c_str()); what = "custom"; } int status = ::ares_get_servers_ports(ares_, &name_servers); if (status == ARES_SUCCESS) { for (auto ns = name_servers; ns != nullptr; ns = ns->next) if (endpoint{ns->family, &ns->addr, static_cast(ns->udp_port)}.format_to(nscsv, endpoint::fmt_default | endpoint::fmt_no_local)) nscsv.push_back(','); if (!nscsv.empty()) // if no valid name server, use predefined fallback dns YASIO_KLOGI("[c-ares] use %s dns: %s", what, nscsv.c_str()); else { status = ::ares_set_servers_csv(ares_, YASIO_FALLBACK_NAME_SERVERS); if (status == 0) YASIO_KLOGW("[c-ares] set fallback dns: '%s' succeed", YASIO_FALLBACK_NAME_SERVERS); else YASIO_KLOGE("[c-ares] set fallback dns: '%s' failed, detail: %s", YASIO_FALLBACK_NAME_SERVERS, ::ares_strerror(status)); } ::ares_free_data(name_servers); } } void io_service::destroy_ares_channel() { if (ares_ != nullptr) { ::ares_cancel(this->ares_); ::ares_destroy(this->ares_); this->ares_ = nullptr; } } #endif void io_service::do_accept(io_channel* ctx) { // channel is server cleanup_channel(ctx); ip::endpoint ep; if (!yasio__testbits(ctx->properties_, YCM_UDS)) { // server: don't need resolve, don't use remote_eps_ auto ifaddr = ctx->remote_host_.empty() ? YASIO_ADDR_ANY(local_address_family()) : ctx->remote_host_.c_str(); ep.as_in(ifaddr, ctx->remote_port_); } #if defined(YASIO_ENABLE_UDS) && YASIO__HAS_UDS else { ep.as_un(ctx->remote_host_.c_str()); ::unlink(ctx->remote_host_.c_str()); } #endif int error = -1; io_base::error_stage where = io_base::error_stage::NONE; do { xxsocket::set_last_errno(0); if (!ctx->socket_->popen(ep.af(), ctx->socktype_)) { where = io_base::error_stage::OPEN_SOCKET; break; } if (yasio__testbits(ctx->properties_, YCF_REUSEADDR)) ctx->socket_->reuse_address(true); if (yasio__testbits(ctx->properties_, YCF_EXCLUSIVEADDRUSE)) ctx->socket_->exclusive_address(false); if (ctx->socket_->bind(ep) != 0) { where = io_base::error_stage::BIND_SOCKET; break; } if (yasio__testbits(ctx->properties_, YCM_TCP) && ctx->socket_->listen(YASIO_SOMAXCONN) != 0) { where = io_base::error_stage::LISTEN_SOCKET; break; } ctx->state_ = io_base::state::OPENED; if (yasio__testbits(ctx->properties_, YCM_UDP)) { if (yasio__testbits(ctx->properties_, YCPF_MCAST)) ctx->join_multicast_group(); ctx->buffer_.resize(YASIO_INET_BUFFER_SIZE); } io_watcher_.mod_event(ctx->socket_->native_handle(), socket_event::read, 0); YASIO_KLOGI("[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); if (error < 0) { error = xxsocket::get_last_errno(); YASIO_KLOGE("[index: %d] open server failed during stage %d, ec=%d, detail:%s", where, ctx->index_, error, io_service::strerror(error)); ctx->socket_->close(); ctx->state_ = io_base::state::CLOSED; } #if defined(YASIO_ENABLE_PASSIVE_EVENT) this->fire_event(ctx->index_, YEK_ON_OPEN, error, ctx, 1); #endif } void io_service::do_accept_completion(io_channel* ctx) { if (ctx->state_ == io_base::state::OPENED) { int error = 0; if (io_watcher_.is_ready(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)) { socket_native_type sockfd{invalid_socket}; error = ctx->socket_->paccept(sockfd); if (error == 0) handle_connect_succeed(ctx, std::make_shared(sockfd)); else // The non-blocking tcp accept failed can be ignored. YASIO_KLOGV("[index: %d] socket.fd=%d, accept failed, ec=%u", ctx->index_, (int)ctx->socket_->native_handle(), error); } else // YCM_UDP { ip::endpoint peer; int n = ctx->socket_->recvfrom(&ctx->buffer_.front(), static_cast(ctx->buffer_.size()), peer); if (n > 0) { YASIO_KLOGV("[index: %d] recvfrom peer: %s succeed.", ctx->index_, peer.to_string().c_str()); auto transport = static_cast(do_dgram_accept(ctx, peer, error)); if (transport) { if (transport->handle_input(ctx->buffer_.data(), n, error, this->wait_duration_) < 0) { transport->error_ = error; close(transport); } } else YASIO_KLOGE("[index: %d] do_dgram_accept failed, ec=%d, detail:%s", ctx->index_, error, this->strerror(error)); } else if (n < 0) { error = xxsocket::get_last_errno(); if (!xxsocket::not_recv_error(error)) YASIO_KLOGE("[index: %d] recvfrom failed, ec=%d, detail:%s", ctx->index_, error, this->strerror(error)); } } } } } transport_handle_t io_service::do_dgram_accept(io_channel* ctx, const ip::endpoint& peer, int& error) { /* Because Bind() the client socket to the socket address of the listening socket. On Linux this essentially passes the responsibility for receiving data for the client session from the well-known listening socket, to the newly allocated client socket. It is important to note that this behavior is not the same on other platforms, like Windows (unfortunately), detail see: https://blog.grijjy.com/2018/08/29/creating-high-performance-udp-servers-on-windows-and-linux https://cloud.tencent.com/developer/article/1004555 So we emulate thus by ourself, don't care the performance, just a workaround implementation. Notes: a. for win32: we check exists udp clients by ourself, and only write operation can be perform on transports, the read event still routed by channel. b. for non-win32 multicast: same with win32, because the kernel can't route same udp peer as 1 transport when the peer always sendto multicast address. */ const bool user_route = !YASIO__UDP_KROUTE || yasio__testbits(ctx->properties_, YCPF_MCAST); if (user_route) { auto it = yasio__find_if(this->transports_, [&peer](const io_transport* transport) { using namespace std; return yasio__testbits(transport->ctx_->properties_, YCM_UDP) && static_cast(transport)->remote_endpoint() == peer; }); if (it != this->transports_.end()) return *it; } auto new_sock = std::make_shared(); if (new_sock->popen(peer.af(), SOCK_DGRAM)) { if (yasio__testbits(ctx->properties_, YCF_REUSEADDR)) new_sock->reuse_address(true); if (yasio__testbits(ctx->properties_, YCF_EXCLUSIVEADDRUSE)) new_sock->exclusive_address(false); if (new_sock->bind(YASIO_ADDR_ANY(peer.af()), ctx->remote_port_) == 0) { auto transport = static_cast(allocate_transport(ctx, std::move(new_sock))); // We always establish 4 tuple with clients transport->confgure_remote(peer); if (user_route) active_transport(transport); else handle_connect_succeed(transport); return transport; } } // unhandled, get error from system. error = xxsocket::get_last_errno(); return nullptr; } void io_service::handle_connect_succeed(transport_handle_t transport) { auto ctx = transport->ctx_; ctx->clear_mutable_flags(); ctx->error_ = 0; // clear errno, value may be EINPROGRESS auto& connection = transport->socket_; if (yasio__testbits(ctx->properties_, YCM_CLIENT)) { // Reset client channel bytes transferred when a new connection established ctx->bytes_transferred_ = 0; ctx->state_ = io_base::state::OPENED; if (yasio__testbits(ctx->properties_, YCM_UDP)) static_cast(transport)->confgure_remote(ctx->remote_eps_[0]); } else io_watcher_.mod_event(connection->native_handle(), socket_event::read, 0); if (yasio__testbits(ctx->properties_, YCM_TCP)) { // apply tcp keepalive options if (options_.tcp_keepalive_.onoff) connection->set_keepalive(options_.tcp_keepalive_.onoff, options_.tcp_keepalive_.idle, options_.tcp_keepalive_.interval, options_.tcp_keepalive_.probs); } active_transport(transport); } void io_service::active_transport(transport_handle_t t) { auto ctx = t->ctx_; auto& s = t->socket_; this->transports_.push_back(t); if (!yasio__testbits(ctx->properties_, YCM_SSL)) { YASIO__UNUSED_PARAM(s); YASIO_KLOGV("[index: %d] sndbuf=%d, rcvbuf=%d", ctx->index_, s->get_optval(SOL_SOCKET, SO_SNDBUF), s->get_optval(SOL_SOCKET, SO_RCVBUF)); YASIO_KLOGD("[index: %d] the connection #%u <%s> --> <%s> is established.", ctx->index_, t->id_, t->local_endpoint().to_string().c_str(), t->remote_endpoint().to_string().c_str()); this->fire_event(ctx->index_, YEK_ON_OPEN, 0, t); } else if (yasio__testbits(ctx->properties_, YCM_CLIENT)) this->wakeup(); } transport_handle_t io_service::allocate_transport(io_channel* ctx, xxsocket_ptr&& s) { transport_handle_t transport; void* vp; if (!tpool_.empty()) { // allocate from pool vp = tpool_.back(); tpool_.pop_back(); } else vp = ::operator new(yasio__max_tsize); do { if (yasio__testbits(ctx->properties_, YCM_TCP)) { // tcp like transport #if defined(YASIO_SSL_BACKEND) if (yasio__testbits(ctx->properties_, YCM_SSL)) { transport = new (vp) io_transport_ssl(ctx, std::forward(s)); break; } #endif transport = new (vp) io_transport_tcp(ctx, std::forward(s)); } else // udp like transport { #if defined(YASIO_ENABLE_KCP) if (yasio__testbits(ctx->properties_, YCM_KCP)) { transport = new (vp) io_transport_kcp(ctx, std::forward(s)); break; } #endif transport = new (vp) io_transport_udp(ctx, std::forward(s)); } } while (false); transport->set_primitives(); return transport; } void io_service::deallocate_transport(transport_handle_t t) { if (t->is_valid()) { yasio::invoke_dtor(t); this->tpool_.push_back(t); } } void io_service::handle_connect_failed(io_channel* ctx, int error) { cleanup_channel(ctx); YASIO_KLOGE("[index: %d] connect server %s failed, ec=%d, detail:%s", ctx->index_, ctx->format_destination().c_str(), error, io_service::strerror(error)); fire_event(ctx->index_, YEK_ON_OPEN, error, ctx); } bool io_service::do_read(transport_handle_t transport) { bool ret = false; do { if (!transport->socket_->is_open()) break; int error = 0; int revent = io_watcher_.is_ready(transport->socket_->native_handle(), socket_event::read | socket_event::error); int n = transport->do_read(revent, error, this->wait_duration_); if (n >= 0) { if (!options_.forward_packet_) { YASIO_KLOGV("[index: %d] do_read status ok, bytes transferred: %d, buffer used: %d", transport->cindex(), n, n + transport->offset_); if (transport->expected_size_ == -1) { // decode length int length = transport->ctx_->decode_len_(transport->buffer_, transport->offset_ + n); if (length > 0) { int bytes_to_strip = ::yasio::clamp(transport->ctx_->uparams_.initial_bytes_to_strip, 0, length - 1); transport->expected_size_ = length; transport->expected_packet_.reserve((std::min)(length - bytes_to_strip, YASIO_MAX_PDU_BUFFER_SIZE)); // #perfomance, avoid memory reallocte. unpack(transport, transport->expected_size_, n, bytes_to_strip); } else if (length == 0) // header insufficient, wait readfd ready at next event frame. transport->offset_ += n; else { transport->set_last_errno(yasio::errc::invalid_packet, yasio::io_base::error_stage::READ); break; } } else // process incompleted pdu unpack(transport, transport->expected_size_ - static_cast(transport->expected_packet_.size()), n, 0); } else if (n > 0) { // forward packet, don't perform unpack, it's useful for implement streaming based protocol, like http, websocket and ... this->forward_packet(transport->cindex(), io_packet_view{transport->buffer_, n}, transport); } } else { // n < 0, regard as connection should close transport->set_last_errno(error, yasio::io_base::error_stage::READ); break; } ret = true; } while (false); return ret; } void io_service::unpack(transport_handle_t transport, int bytes_expected, int bytes_transferred, int bytes_to_strip) { auto& offset = transport->offset_; auto bytes_available = bytes_transferred + offset; auto& pkt = transport->expected_packet_; pkt.insert(pkt.end(), transport->buffer_ + bytes_to_strip, transport->buffer_ + (std::min)(bytes_expected, bytes_available)); // set 'offset' to bytes of remain buffer offset = bytes_available - bytes_expected; if (offset >= 0) { /* pdu received properly */ if (offset > 0) { /* move remain data to head of buffer and hold 'offset'. */ ::memmove(transport->buffer_, transport->buffer_ + bytes_expected, offset); this->wait_duration_ = 0; } // move properly pdu to ready queue, the other thread who care about will retrieve it. YASIO_KLOGV("[index: %d] received a properly packet from peer, packet size:%d", transport->cindex(), transport->expected_size_); this->fire_event(transport->cindex(), transport->fetch_packet(), transport); } else /* all buffer consumed, set 'offset' to ZERO, pdu incomplete, continue recv remain data. */ offset = 0; } highp_timer_ptr io_service::schedule(const std::chrono::microseconds& duration, timer_cb_t cb) { auto timer = std::make_shared(*this); timer->expires_from_now(duration); /*!important, hold on `timer` by lambda expression */ #if YASIO__HAS_CXX14 timer->async_wait([timer, cb = std::move(cb)](io_service& service) { return cb(service); }); #else timer->async_wait([timer, cb](io_service& service) { return cb(service); }); #endif return timer; } void io_service::schedule_timer(highp_timer* timer_ctl, timer_cb_t&& timer_cb) { if (timer_ctl == nullptr) return; std::lock_guard lck(this->timer_queue_mtx_); auto timer_it = this->find_timer(timer_ctl); if (timer_it == timer_queue_.end()) this->timer_queue_.emplace_back(timer_ctl, std::move(timer_cb)); else // always replace timer_cb timer_it->second = std::move(timer_cb); this->sort_timers(); // If the timer is earliest, wakup if (timer_ctl == this->timer_queue_.back().first) this->wakeup(); } void io_service::remove_timer(highp_timer* timer) { std::lock_guard lck(this->timer_queue_mtx_); auto iter = this->find_timer(timer); if (iter != timer_queue_.end()) { timer_queue_.erase(iter); if (!timer_queue_.empty()) { this->sort_timers(); this->wakeup(); } } } bool io_service::open_internal(io_channel* ctx) { if (ctx->state_ == io_base::state::CONNECTING || ctx->state_ == io_base::state::RESOLVING) { YASIO_KLOGD("[index: %d] the channel open operation is in progress!", ctx->index_); return false; } yasio__clearbits(ctx->opmask_, YOPM_CLOSE); yasio__setbits(ctx->opmask_, YOPM_OPEN); ++ctx->connect_id_; this->channel_ops_mtx_.lock(); if (yasio__find(this->channel_ops_, ctx) == this->channel_ops_.end()) this->channel_ops_.push_back(ctx); this->channel_ops_mtx_.unlock(); this->wakeup(); return true; } bool io_service::close_internal(io_channel* ctx) { yasio__clearbits(ctx->opmask_, YOPM_OPEN); if (ctx->socket_->is_open()) { yasio__setbits(ctx->opmask_, YOPM_CLOSE); return true; } return false; } void io_service::process_timers() { this->update_time(); if (this->timer_queue_.empty()) return; std::lock_guard lck(this->timer_queue_mtx_); unsigned int n = 0; // the count expired loop timers while (!this->timer_queue_.empty()) { auto timer_ctl = timer_queue_.back().first; if (timer_ctl->expired()) { // fetch timer auto timer_impl = std::move(timer_queue_.back()); timer_queue_.pop_back(); if (!timer_impl.second(*this)) { // reschedule if the timer want wait again timer_ctl->expires_from_now(); timer_queue_.push_back(std::move(timer_impl)); ++n; } } else break; } if (n) this->sort_timers(); } void io_service::process_deferred_events() { if (!options_.no_dispatch_ && dispatch() > 0) this->wait_duration_ = 0; } highp_time_t io_service::get_timeout(highp_time_t usec) { this->wait_duration_ = yasio__max_wait_usec; // Reset next wait duration per frame if (this->timer_queue_.empty()) return usec; std::lock_guard lck(this->timer_queue_mtx_); if (!this->timer_queue_.empty()) { // microseconds auto duration = timer_queue_.back().first->wait_duration(); if (std::chrono::microseconds(usec) > duration) usec = duration.count(); } return usec; } bool io_service::cleanup_channel(io_channel* ctx, bool clear_mask) { ctx->clear_mutable_flags(); bool bret = cleanup_io(ctx, clear_mask); #if defined(YASIO_ENABLE_PASSIVE_EVENT) if (bret && yasio__testbits(ctx->properties_, YCM_SERVER)) this->fire_event(ctx->index_, YEK_ON_CLOSE, 0, ctx, 1); #endif return bret; } bool io_service::cleanup_io(io_base* obj, bool clear_mask) { obj->error_ = 0; obj->state_ = io_base::state::CLOSED; if (clear_mask) obj->opmask_ = 0; if (obj->socket_->is_open()) { io_watcher_.mod_event(obj->socket_->native_handle(), 0, socket_event::readwrite); obj->socket_->close(); return true; } return false; } int io_service::do_resolve(io_channel* ctx) { if (yasio__testbits(ctx->properties_, YCPF_HOST_DIRTY)) { yasio__clearbits(ctx->properties_, YCPF_HOST_DIRTY | YCPF_NEEDS_QUERY); ctx->remote_eps_.clear(); ip::endpoint ep; #if defined(YASIO_ENABLE_UDS) && YASIO__HAS_UDS if (yasio__testbits(ctx->properties_, YCM_UDS)) { ep.as_un(ctx->remote_host_.c_str()); ctx->remote_eps_.push_back(ep); return 0; } #endif if (ep.as_in(ctx->remote_host_.c_str(), ctx->remote_port_)) ctx->remote_eps_.push_back(ep); else yasio__setbits(ctx->properties_, YCPF_NEEDS_QUERY); } if (yasio__testbits(ctx->properties_, YCPF_PORT_DIRTY)) { yasio__clearbits(ctx->properties_, YCPF_PORT_DIRTY); for (auto& ep : ctx->remote_eps_) ep.port(ctx->remote_port_); } if (!ctx->remote_eps_.empty()) { if (!yasio__testbits(ctx->properties_, YCPF_NEEDS_QUERY)) return 0; // remote host is IP address, no needs to query update_dns_status(); // will reset our resolved address expire time if (ctx->error_ == EINPROGRESS) return 0; // queried address not consumed by this connect flow if (!address_expired(ctx)) return 0; // queried address not exipred ctx->remote_eps_.clear(); } if (!ctx->remote_host_.empty()) { if (!ctx->error_) start_query(ctx); } else ctx->error_ = yasio::errc::no_available_address; return -1; } void io_service::start_query(io_channel* ctx) { ctx->set_last_errno(EINPROGRESS); YASIO_KLOGD("[index: %d] start query %s...", ctx->index_, ctx->remote_host_.c_str()); #if defined(YASIO_ENABLE_ARES_PROFILER) ctx->query_start_time_ = highp_clock(); #endif #if !defined(YASIO_USE_CARES) // init async name query thread state auto resolving_host = ctx->remote_host_; auto resolving_port = ctx->remote_port_; std::weak_ptr weak_mutex = life_mutex_; std::weak_ptr life_token = life_token_; std::thread async_resolv_thread([this, life_token, weak_mutex, resolving_host, resolving_port, ctx] { // check life token if (life_token.use_count() < 1) return; // preform blocking resolving safe std::vector remote_eps; int error = options_.resolv_(remote_eps, resolving_host.c_str(), resolving_port); // lock perform update dns state of the channel auto pmtx = weak_mutex.lock(); if (!pmtx) return; cxx17::shared_lock lck(*pmtx); // check life token again, when io_service cleanup done, life_token's use_count will be 0, // otherwise, we can safe to do follow assignments. if (life_token.use_count() < 1) return; if (error == 0) { ctx->remote_eps_ = std::move(remote_eps); ctx->query_success_time_ = highp_clock(); # if defined(YASIO_ENABLE_ARES_PROFILER) YASIO_KLOGD("[index: %d] query %s succeed, cost: %g(ms)", ctx->index_, ctx->remote_host_.c_str(), (ctx->query_success_time_ - ctx->query_start_time_) / 1000.0); # endif } else { ctx->set_last_errno(yasio::errc::resolve_host_failed); YASIO_KLOGE("[index: %d] query %s failed, ec=%d, detail:%s", ctx->index_, ctx->remote_host_.c_str(), error, xxsocket::gai_strerror(error)); } this->wakeup(); }); async_resolv_thread.detach(); #else ares_addrinfo_hints hint; memset(&hint, 0x0, sizeof(hint)); hint.ai_family = local_address_family(); char sport[sizeof "65535"] = {'\0'}; const char* service = nullptr; if (ctx->remote_port_ > 0) { snprintf(sport, sizeof(sport), "%u", ctx->remote_port_); service = sport; } ares_work_started(); ::ares_getaddrinfo(this->ares_, ctx->remote_host_.c_str(), service, &hint, io_service::ares_getaddrinfo_cb, ctx); #endif } void io_service::update_dns_status() { if (this->options_.dns_dirty_) { this->options_.dns_dirty_ = false; #if defined(YASIO_USE_CARES) recreate_ares_channel(); #endif for (auto channel : this->channels_) channel->query_success_time_ = 0; } } int io_service::resolve(std::vector& endpoints, const char* hostname, unsigned short port) { if (yasio__testbits(this->ipsv_, ipsv_ipv4)) return xxsocket::resolve_v4(endpoints, hostname, port); else if (yasio__testbits(this->ipsv_, ipsv_ipv6)) // localhost is IPv6_only network return xxsocket::resolve_v6(endpoints, hostname, port) != 0 ? xxsocket::resolve_v4to6(endpoints, hostname, port) : 0; return -1; } void io_service::wakeup() { io_watcher_.wakeup(); } const char* io_service::strerror(int error) { switch (error) { case 0: return "No error."; case yasio::errc::resolve_host_failed: return "Resolve host failed!"; case yasio::errc::no_available_address: return "No available address!"; case yasio::errc::shutdown_by_localhost: return "An existing connection was shutdown by local host!"; case yasio::errc::invalid_packet: return "Invalid packet!"; case yasio::errc::ssl_handshake_failed: return "SSL handshake failed!"; case yasio::errc::ssl_write_failed: return "SSL write failed!"; case yasio::errc::ssl_read_failed: return "SSL read failed!"; case yasio::errc::read_timeout: return "The remote host did not respond after a period of time."; case yasio::errc::eof: return "End of file."; case -1: return "Unknown error!"; default: return xxsocket::strerror(error); } } void io_service::set_option(int opt, ...) { va_list ap; va_start(ap, opt); set_option_internal(opt, ap); va_end(ap); } void io_service::set_option_internal(int opt, va_list ap) // lgtm [cpp/poorly-documented-function] { switch (opt) { case YOPT_S_NO_DISPATCH: options_.no_dispatch_ = !!va_arg(ap, int); break; case YOPT_S_TCP_KEEPALIVE: options_.tcp_keepalive_.onoff = 1; options_.tcp_keepalive_.idle = va_arg(ap, int); options_.tcp_keepalive_.interval = va_arg(ap, int); options_.tcp_keepalive_.probs = va_arg(ap, int); break; case YOPT_S_RESOLV_FN: options_.resolv_ = *va_arg(ap, resolv_fn_t*); break; case YOPT_S_PRINT_FN: { auto ncb = *va_arg(ap, print_fn_t*); if (ncb) options_.print_ = [=](int, const char* msg) { ncb(msg); }; else options_.print_ = nullptr; } break; case YOPT_S_PRINT_FN2: options_.print_ = *va_arg(ap, print_fn2_t*); break; case YOPT_S_NO_NEW_THREAD: this->options_.no_new_thread_ = !!va_arg(ap, int); break; #if defined(YASIO_SSL_BACKEND) case YOPT_S_SSL_CACERT: this->options_.cafile_ = va_arg(ap, const char*); break; #endif case YOPT_S_CONNECT_TIMEOUT: options_.connect_timeout_ = static_cast(va_arg(ap, int)) * std::micro::den; break; case YOPT_S_CONNECT_TIMEOUTMS: options_.connect_timeout_ = static_cast(va_arg(ap, int)) * std::milli::den; break; case YOPT_S_DNS_CACHE_TIMEOUT: options_.dns_cache_timeout_ = static_cast(va_arg(ap, int)) * std::micro::den; break; case YOPT_S_DNS_CACHE_TIMEOUTMS: options_.dns_cache_timeout_ = static_cast(va_arg(ap, int)) * std::milli::den; break; case YOPT_S_DNS_QUERIES_TIMEOUT: options_.dns_queries_timeout_ = static_cast(va_arg(ap, int)) * std::micro::den; break; case YOPT_S_DNS_QUERIES_TIMEOUTMS: options_.dns_queries_timeout_ = static_cast(va_arg(ap, int)) * std::milli::den; break; case YOPT_S_DNS_QUERIES_TRIES: options_.dns_queries_tries_ = va_arg(ap, int); break; case YOPT_S_DNS_DIRTY: options_.dns_dirty_ = true; break; #if defined(YASIO_USE_CARES) case YOPT_S_DNS_LIST: options_.name_servers_ = va_arg(ap, const char*); options_.dns_dirty_ = true; break; #endif case YOPT_S_EVENT_CB: options_.on_event_ = *va_arg(ap, event_cb_t*); break; case YOPT_S_DEFER_EVENT_CB: options_.on_defer_event_ = *va_arg(ap, defer_event_cb_t*); break; case YOPT_S_FORWARD_PACKET: options_.forward_packet_ = !!va_arg(ap, int); break; #if defined(YASIO_SSL_BACKEND) case YOPT_S_SSL_CERT: options_.crtfile_ = va_arg(ap, const char*); options_.keyfile_ = va_arg(ap, const char*); break; #endif case YOPT_C_UNPACK_PARAMS: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { channel->uparams_.max_frame_length = va_arg(ap, int); channel->uparams_.length_field_offset = va_arg(ap, int); channel->uparams_.length_field_length = yasio::clamp(va_arg(ap, int), YASIO_SSIZEOF(int8_t), YASIO_SSIZEOF(int)); channel->uparams_.length_adjustment = va_arg(ap, int); } break; } case YOPT_C_UNPACK_STRIP: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->uparams_.initial_bytes_to_strip = yasio::clamp(va_arg(ap, int), 0, YASIO_UNPACK_MAX_STRIP); break; } case YOPT_C_UNPACK_NO_BSWAP: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->uparams_.no_bswap = va_arg(ap, int); break; } case YOPT_C_LFBFD_FN: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->decode_len_ = *va_arg(ap, decode_len_fn_t*); break; } case YOPT_C_LOCAL_HOST: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->local_host_ = va_arg(ap, const char*); break; } case YOPT_C_LOCAL_PORT: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->local_port_ = (u_short)va_arg(ap, int); break; } case YOPT_C_LOCAL_ENDPOINT: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { channel->local_host_ = va_arg(ap, const char*); channel->local_port_ = ((u_short)va_arg(ap, int)); } break; } case YOPT_C_REMOTE_HOST: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->set_host(va_arg(ap, const char*)); break; } case YOPT_C_REMOTE_PORT: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->set_port((u_short)va_arg(ap, int)); break; } case YOPT_C_REMOTE_ENDPOINT: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { channel->set_host(va_arg(ap, const char*)); channel->set_port((u_short)va_arg(ap, int)); } break; } case YOPT_C_MCAST_IF: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { const char* ifaddr = va_arg(ap, const char*); if (ifaddr) channel->multiif_.as_in(ifaddr, (unsigned short)0); } break; } case YOPT_C_ENABLE_MCAST: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { const char* addr = va_arg(ap, const char*); int loopback = va_arg(ap, int); channel->enable_multicast(addr, loopback); if (channel->socket_->is_open()) channel->join_multicast_group(); } break; } case YOPT_C_DISABLE_MCAST: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->disable_multicast(); break; } case YOPT_C_MOD_FLAGS: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { yasio__setbits(channel->properties_, (uint32_t)va_arg(ap, int)); yasio__clearbits(channel->properties_, (uint32_t)va_arg(ap, int)); } break; } #if defined(YASIO_ENABLE_KCP) case YOPT_C_KCP_CONV: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->kcp_options().kcp_conv_ = va_arg(ap, int); break; } case YOPT_C_KCP_NODELAY: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { // nodelay:int, interval:int, resend:int, nc:int. channel->kcp_options().kcp_nodelay_ = va_arg(ap, int); channel->kcp_options().kcp_interval_ = va_arg(ap, int); channel->kcp_options().kcp_resend_ = va_arg(ap, int); channel->kcp_options().kcp_ncwnd_ = va_arg(ap, int); } break; } case YOPT_C_KCP_WINDOW_SIZE: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) { channel->kcp_options().kcp_sndwnd_ = va_arg(ap, int); channel->kcp_options().kcp_rcvwnd_ = va_arg(ap, int); } break; } case YOPT_C_KCP_MTU: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->kcp_options().kcp_mtu_ = va_arg(ap, int); break; } case YOPT_C_KCP_RTO_MIN: { auto channel = channel_at(static_cast(va_arg(ap, int))); if (channel) channel->kcp_options().kcp_minrto_ = va_arg(ap, int); break; } #endif case YOPT_T_CONNECT: { auto transport = va_arg(ap, transport_handle_t); if (transport && transport->is_open() && (transport->ctx_->properties_ & 0xff) == YCK_UDP_CLIENT) static_cast(transport)->connect(); break; } case YOPT_T_DISCONNECT: { auto transport = va_arg(ap, transport_handle_t); if (transport && transport->is_open() && (transport->ctx_->properties_ & 0xff) == YCK_UDP_CLIENT) static_cast(transport)->disconnect(); break; } case YOPT_B_SOCKOPT: { auto obj = va_arg(ap, io_base*); if (obj && obj->socket_ && obj->socket_->is_open()) { auto optlevel = va_arg(ap, int); auto optname = va_arg(ap, int); auto optval = va_arg(ap, void*); auto optlen = va_arg(ap, int); obj->socket_->set_optval(optlevel, optname, optval, optlen); } break; } } } } // namespace inet } // namespace yasio #if defined(_MSC_VER) # pragma warning(pop) #endif #endif