#include "ip.h" #include "socket.h" #include "address.h" #include "pollerimpl.h" #include "iovec.h" #include #include #if defined(_unix_) #include #include #include #include #include #include #include #include #endif #if defined(_freebsd_) #include #define ACCEPT_FILTER_MOD #include #endif #if defined(_win_) #include #include #include #include #include #endif #include #include #include #include #include #include #include #include #include #include using namespace NAddr; #if defined(_win_) int inet_aton(const char* cp, struct in_addr* inp) { sockaddr_in addr; addr.sin_family = AF_INET; int psz = sizeof(addr); if (0 == WSAStringToAddress((char*)cp, AF_INET, nullptr, (LPSOCKADDR)&addr, &psz)) { memcpy(inp, &addr.sin_addr, sizeof(in_addr)); return 1; } return 0; } #if (_WIN32_WINNT < 0x0600) const char* inet_ntop(int af, const void* src, char* dst, socklen_t size) { if (af != AF_INET) { errno = EINVAL; return 0; } const ui8* ia = (ui8*)src; if (snprintf(dst, size, "%u.%u.%u.%u", ia[0], ia[1], ia[2], ia[3]) >= (int)size) { errno = ENOSPC; return 0; } return dst; } struct evpair { int event; int winevent; }; static const evpair evpairs_to_win[] = { {POLLIN, FD_READ | FD_CLOSE | FD_ACCEPT}, {POLLRDNORM, FD_READ | FD_CLOSE | FD_ACCEPT}, {POLLRDBAND, -1}, {POLLPRI, -1}, {POLLOUT, FD_WRITE | FD_CLOSE}, {POLLWRNORM, FD_WRITE | FD_CLOSE}, {POLLWRBAND, -1}, {POLLERR, 0}, {POLLHUP, 0}, {POLLNVAL, 0}}; static const size_t nevpairs_to_win = sizeof(evpairs_to_win) / sizeof(evpairs_to_win[0]); static const evpair evpairs_to_unix[] = { {FD_ACCEPT, POLLIN | POLLRDNORM}, {FD_READ, POLLIN | POLLRDNORM}, {FD_WRITE, POLLOUT | POLLWRNORM}, {FD_CLOSE, POLLHUP}, }; static const size_t nevpairs_to_unix = sizeof(evpairs_to_unix) / sizeof(evpairs_to_unix[0]); static int convert_events(int events, const evpair* evpairs, size_t nevpairs, bool ignoreUnknown) noexcept { int result = 0; for (size_t i = 0; i < nevpairs; ++i) { int event = evpairs[i].event; if (events & event) { events ^= event; long winEvent = evpairs[i].winevent; if (winEvent == -1) return -1; if (winEvent == 0) continue; result |= winEvent; } } if (events != 0 && !ignoreUnknown) return -1; return result; } class TWSAEventHolder { private: HANDLE Event; public: inline TWSAEventHolder(HANDLE event) noexcept : Event(event) { } inline ~TWSAEventHolder() { WSACloseEvent(Event); } inline HANDLE Get() noexcept { return Event; } }; int poll(struct pollfd fds[], nfds_t nfds, int timeout) noexcept { HANDLE rawEvent = WSACreateEvent(); if (rawEvent == WSA_INVALID_EVENT) { errno = EIO; return -1; } TWSAEventHolder event(rawEvent); int checked_sockets = 0; for (pollfd* fd = fds; fd < fds + nfds; ++fd) { int win_events = convert_events(fd->events, evpairs_to_win, nevpairs_to_win, false); if (win_events == -1) { errno = EINVAL; return -1; } fd->revents = 0; if (WSAEventSelect(fd->fd, event.Get(), win_events)) { int error = WSAGetLastError(); if (error == WSAEINVAL || error == WSAENOTSOCK) { fd->revents = POLLNVAL; ++checked_sockets; } else { errno = EIO; return -1; } } fd_set readfds; fd_set writefds; struct timeval timeout = {0, 0}; FD_ZERO(&readfds); FD_ZERO(&writefds); if (fd->events & POLLIN) { FD_SET(fd->fd, &readfds); } if (fd->events & POLLOUT) { FD_SET(fd->fd, &writefds); } int error = select(0, &readfds, &writefds, nullptr, &timeout); if (error > 0) { if (FD_ISSET(fd->fd, &readfds)) { fd->revents |= POLLIN; } if (FD_ISSET(fd->fd, &writefds)) { fd->revents |= POLLOUT; } ++checked_sockets; } } if (checked_sockets > 0) { // returns without wait since we already have sockets in desired conditions return checked_sockets; } HANDLE events[] = {event.Get()}; DWORD wait_result = WSAWaitForMultipleEvents(1, events, TRUE, timeout, FALSE); if (wait_result == WSA_WAIT_TIMEOUT) return 0; else if (wait_result == WSA_WAIT_EVENT_0) { for (pollfd* fd = fds; fd < fds + nfds; ++fd) { if (fd->revents == POLLNVAL) continue; WSANETWORKEVENTS network_events; if (WSAEnumNetworkEvents(fd->fd, event.Get(), &network_events)) { errno = EIO; return -1; } fd->revents = 0; for (int i = 0; i < FD_MAX_EVENTS; ++i) { if ((network_events.lNetworkEvents & (1 << i)) != 0 && network_events.iErrorCode[i]) { fd->revents = POLLERR; break; } } if (fd->revents == POLLERR) continue; if (network_events.lNetworkEvents) { fd->revents = static_cast(convert_events(network_events.lNetworkEvents, evpairs_to_unix, nevpairs_to_unix, true)); if (fd->revents & POLLHUP) { fd->revents &= POLLHUP | POLLIN | POLLRDNORM; } } } int chanded_sockets = 0; for (pollfd* fd = fds; fd < fds + nfds; ++fd) if (fd->revents != 0) ++chanded_sockets; return chanded_sockets; } else { errno = EIO; return -1; } } #endif #endif bool GetRemoteAddr(SOCKET Socket, char* str, socklen_t size) { if (!size) { return false; } TOpaqueAddr addr; if (getpeername(Socket, addr.MutableAddr(), addr.LenPtr()) != 0) { return false; } try { TMemoryOutput out(str, size - 1); PrintHost(out, addr); *out.Buf() = 0; return true; } catch (...) { // ¯\_(ツ)_/¯ } return false; } void SetSocketTimeout(SOCKET s, long timeout) { SetSocketTimeout(s, timeout, 0); } void SetSocketTimeout(SOCKET s, long sec, long msec) { #ifdef SO_SNDTIMEO #ifdef _darwin_ const timeval timeout = {sec, (__darwin_suseconds_t)msec * 1000}; #elif defined(_emscripten_) const timeval timeout = {sec, static_cast(msec * 1000)}; #elif defined(_unix_) const timeval timeout = {sec, msec * 1000}; #else const int timeout = sec * 1000 + msec; #endif CheckedSetSockOpt(s, SOL_SOCKET, SO_RCVTIMEO, timeout, "recv timeout"); CheckedSetSockOpt(s, SOL_SOCKET, SO_SNDTIMEO, timeout, "send timeout"); #endif } void SetLinger(SOCKET s, bool on, unsigned len) { #ifdef SO_LINGER struct linger l = {on, (u_short)len}; CheckedSetSockOpt(s, SOL_SOCKET, SO_LINGER, l, "linger"); #endif } void SetZeroLinger(SOCKET s) { SetLinger(s, 1, 0); } void SetKeepAlive(SOCKET s, bool value) { CheckedSetSockOpt(s, SOL_SOCKET, SO_KEEPALIVE, (int)value, "keepalive"); } void SetOutputBuffer(SOCKET s, unsigned value) { CheckedSetSockOpt(s, SOL_SOCKET, SO_SNDBUF, value, "output buffer"); } void SetInputBuffer(SOCKET s, unsigned value) { CheckedSetSockOpt(s, SOL_SOCKET, SO_RCVBUF, value, "input buffer"); } void SetReusePort(SOCKET s, bool value) { #if defined(_unix_) CheckedSetSockOpt(s, SOL_SOCKET, SO_REUSEPORT, (int)value, "reuse port"); #else Y_UNUSED(s); Y_UNUSED(value); ythrow TSystemError(ENOSYS) << "SO_REUSEPORT is not available on Windows"; #endif } void SetNoDelay(SOCKET s, bool value) { CheckedSetSockOpt(s, IPPROTO_TCP, TCP_NODELAY, (int)value, "tcp no delay"); } void SetCloseOnExec(SOCKET s, bool value) { #if defined(_unix_) int flags = fcntl(s, F_GETFD); if (flags == -1) { ythrow TSystemError() << "fcntl() failed"; } if (value) { flags |= FD_CLOEXEC; } else { flags &= ~FD_CLOEXEC; } if (fcntl(s, F_SETFD, flags) == -1) { ythrow TSystemError() << "fcntl() failed"; } #else Y_UNUSED(s); Y_UNUSED(value); #endif } size_t GetMaximumSegmentSize(SOCKET s) { #if defined(TCP_MAXSEG) int val; if (GetSockOpt(s, IPPROTO_TCP, TCP_MAXSEG, val) == 0) { return (size_t)val; } #endif /* * probably a good guess... */ return 8192; } size_t GetMaximumTransferUnit(SOCKET /*s*/) { // for someone who'll dare to write it // Linux: there rummored to be IP_MTU getsockopt() request // FreeBSD: request to a socket of type PF_ROUTE // with peer address as a destination argument return 8192; } int GetSocketToS(SOCKET s) { TOpaqueAddr addr; if (getsockname(s, addr.MutableAddr(), addr.LenPtr()) < 0) { ythrow TSystemError() << "getsockname() failed"; } return GetSocketToS(s, &addr); } int GetSocketToS(SOCKET s, const IRemoteAddr* addr) { int result = 0; switch (addr->Addr()->sa_family) { case AF_INET: CheckedGetSockOpt(s, IPPROTO_IP, IP_TOS, result, "tos"); break; case AF_INET6: #ifdef IPV6_TCLASS CheckedGetSockOpt(s, IPPROTO_IPV6, IPV6_TCLASS, result, "tos"); #endif break; } return result; } void SetSocketToS(SOCKET s, const NAddr::IRemoteAddr* addr, int tos) { switch (addr->Addr()->sa_family) { case AF_INET: CheckedSetSockOpt(s, IPPROTO_IP, IP_TOS, tos, "tos"); return; case AF_INET6: #ifdef IPV6_TCLASS CheckedSetSockOpt(s, IPPROTO_IPV6, IPV6_TCLASS, tos, "tos"); return; #endif break; } ythrow yexception() << "SetSocketToS unsupported for family " << addr->Addr()->sa_family; } void SetSocketToS(SOCKET s, int tos) { TOpaqueAddr addr; if (getsockname(s, addr.MutableAddr(), addr.LenPtr()) < 0) { ythrow TSystemError() << "getsockname() failed"; } SetSocketToS(s, &addr, tos); } void SetSocketPriority(SOCKET s, int priority) { #if defined(SO_PRIORITY) CheckedSetSockOpt(s, SOL_SOCKET, SO_PRIORITY, priority, "priority"); #else Y_UNUSED(s); Y_UNUSED(priority); #endif } bool HasLocalAddress(SOCKET socket) { TOpaqueAddr localAddr; if (getsockname(socket, localAddr.MutableAddr(), localAddr.LenPtr()) != 0) { ythrow TSystemError() << "HasLocalAddress: getsockname() failed. "; } if (IsLoopback(localAddr)) { return true; } TOpaqueAddr remoteAddr; if (getpeername(socket, remoteAddr.MutableAddr(), remoteAddr.LenPtr()) != 0) { ythrow TSystemError() << "HasLocalAddress: getpeername() failed. "; } return IsSame(localAddr, remoteAddr); } namespace { #if defined(_linux_) #if !defined(TCP_FASTOPEN) #define TCP_FASTOPEN 23 #endif #endif #if defined(TCP_FASTOPEN) struct TTcpFastOpenFeature { inline TTcpFastOpenFeature() : HasFastOpen_(false) { TSocketHolder tmp(socket(AF_INET, SOCK_STREAM, 0)); int val = 1; int ret = SetSockOpt(tmp, IPPROTO_TCP, TCP_FASTOPEN, val); HasFastOpen_ = (ret == 0); } inline void SetFastOpen(SOCKET s, int qlen) const { if (HasFastOpen_) { CheckedSetSockOpt(s, IPPROTO_TCP, TCP_FASTOPEN, qlen, "setting TCP_FASTOPEN"); } } static inline const TTcpFastOpenFeature* Instance() noexcept { return Singleton(); } bool HasFastOpen_; }; #endif } // namespace void SetTcpFastOpen(SOCKET s, int qlen) { #if defined(TCP_FASTOPEN) TTcpFastOpenFeature::Instance()->SetFastOpen(s, qlen); #else Y_UNUSED(s); Y_UNUSED(qlen); #endif } static bool IsBlocked(int lasterr) noexcept { return lasterr == EAGAIN || lasterr == EWOULDBLOCK; } struct TUnblockingGuard { SOCKET S_; TUnblockingGuard(SOCKET s) : S_(s) { SetNonBlock(S_, true); } ~TUnblockingGuard() { SetNonBlock(S_, false); } }; static int MsgPeek(SOCKET s) { int flags = MSG_PEEK; #if defined(_win_) TUnblockingGuard unblocker(s); Y_UNUSED(unblocker); #else flags |= MSG_DONTWAIT; #endif char c; return recv(s, &c, 1, flags); } bool IsNotSocketClosedByOtherSide(SOCKET s) { return HasSocketDataToRead(s) != ESocketReadStatus::SocketClosed; } ESocketReadStatus HasSocketDataToRead(SOCKET s) { const int r = MsgPeek(s); if (r == -1 && IsBlocked(LastSystemError())) { return ESocketReadStatus::NoData; } if (r > 0) { return ESocketReadStatus::HasData; } return ESocketReadStatus::SocketClosed; } #if defined(_win_) static ssize_t DoSendMsg(SOCKET sock, const struct iovec* iov, int iovcnt) { return writev(sock, iov, iovcnt); } #else static ssize_t DoSendMsg(SOCKET sock, const struct iovec* iov, int iovcnt) { struct msghdr message; Zero(message); message.msg_iov = const_cast(iov); message.msg_iovlen = iovcnt; return sendmsg(sock, &message, MSG_NOSIGNAL); } #endif void TSocketHolder::Close() noexcept { if (Fd_ != INVALID_SOCKET) { bool ok = (closesocket(Fd_) == 0); if (!ok) { // Do not quietly close bad descriptor, // because often it means double close // that is disasterous #ifdef _win_ Y_ABORT_UNLESS(WSAGetLastError() != WSAENOTSOCK, "must not quietly close bad socket descriptor"); #elif defined(_unix_) Y_ABORT_UNLESS(errno != EBADF, "must not quietly close bad descriptor: fd=%d", int(Fd_)); #else #error unsupported platform #endif } Fd_ = INVALID_SOCKET; } } class TSocket::TImpl: public TAtomicRefCount { using TOps = TSocket::TOps; public: inline TImpl(SOCKET fd, TOps* ops) : Fd_(fd) , Ops_(ops) { } inline ~TImpl() = default; inline SOCKET Fd() const noexcept { return Fd_; } inline ssize_t Send(const void* data, size_t len) { return Ops_->Send(Fd_, data, len); } inline ssize_t Recv(void* buf, size_t len) { return Ops_->Recv(Fd_, buf, len); } inline ssize_t SendV(const TPart* parts, size_t count) { return Ops_->SendV(Fd_, parts, count); } inline void Close() { Fd_.Close(); } private: TSocketHolder Fd_; TOps* Ops_; }; template <> void Out(IOutputStream& os, const struct addrinfo* ai) { if (ai->ai_flags & AI_CANONNAME) { os << "`" << ai->ai_canonname << "' "; } os << '['; for (int i = 0; ai; ++i, ai = ai->ai_next) { if (i > 0) { os << ", "; } os << (const IRemoteAddr&)TAddrInfo(ai); } os << ']'; } template <> void Out(IOutputStream& os, struct addrinfo* ai) { Out(os, static_cast(ai)); } template <> void Out(IOutputStream& os, const TNetworkAddress& addr) { os << &*addr.Begin(); } static inline const struct addrinfo* Iterate(const struct addrinfo* addr, const struct addrinfo* addr0, const int sockerr) { if (addr->ai_next) { return addr->ai_next; } ythrow TSystemError(sockerr) << "can not connect to " << addr0; } static inline SOCKET DoConnectImpl(const struct addrinfo* res, const TInstant& deadLine) { const struct addrinfo* addr0 = res; while (res) { TSocketHolder s(socket(res->ai_family, res->ai_socktype, res->ai_protocol)); if (s.Closed()) { res = Iterate(res, addr0, LastSystemError()); continue; } SetNonBlock(s, true); if (connect(s, res->ai_addr, (int)res->ai_addrlen)) { int err = LastSystemError(); if (err == EINPROGRESS || err == EAGAIN || err == EWOULDBLOCK) { /* * must wait */ struct pollfd p = { (SOCKET)s, POLLOUT, 0}; const ssize_t n = PollD(&p, 1, deadLine); /* * timeout occured */ if (n < 0) { ythrow TSystemError(-(int)n) << "can not connect"; } CheckedGetSockOpt(s, SOL_SOCKET, SO_ERROR, err, "socket error"); if (!err) { return s.Release(); } } res = Iterate(res, addr0, err); continue; } return s.Release(); } ythrow yexception() << "something went wrong: nullptr at addrinfo"; } static inline SOCKET DoConnect(const struct addrinfo* res, const TInstant& deadLine) { TSocketHolder ret(DoConnectImpl(res, deadLine)); SetNonBlock(ret, false); return ret.Release(); } static inline ssize_t DoSendV(SOCKET fd, const struct iovec* iov, size_t count) { ssize_t ret = -1; do { ret = DoSendMsg(fd, iov, (int)count); } while (ret == -1 && errno == EINTR); if (ret < 0) { return -LastSystemError(); } return ret; } template struct TSender { using TPart = TSocket::TPart; static inline ssize_t SendV(SOCKET fd, const TPart* parts, size_t count) { return DoSendV(fd, (const iovec*)parts, count); } }; template <> struct TSender { using TPart = TSocket::TPart; static inline ssize_t SendV(SOCKET fd, const TPart* parts, size_t count) { TTempBuf tempbuf(sizeof(struct iovec) * count); struct iovec* iov = (struct iovec*)tempbuf.Data(); for (size_t i = 0; i < count; ++i) { struct iovec& io = iov[i]; const TPart& part = parts[i]; io.iov_base = (char*)part.buf; io.iov_len = part.len; } return DoSendV(fd, iov, count); } }; class TCommonSockOps: public TSocket::TOps { using TPart = TSocket::TPart; public: inline TCommonSockOps() noexcept { } ~TCommonSockOps() override = default; ssize_t Send(SOCKET fd, const void* data, size_t len) override { ssize_t ret = -1; do { ret = send(fd, (const char*)data, (int)len, MSG_NOSIGNAL); } while (ret == -1 && errno == EINTR); if (ret < 0) { return -LastSystemError(); } return ret; } ssize_t Recv(SOCKET fd, void* buf, size_t len) override { ssize_t ret = -1; do { ret = recv(fd, (char*)buf, (int)len, 0); } while (ret == -1 && errno == EINTR); if (ret < 0) { return -LastSystemError(); } return ret; } ssize_t SendV(SOCKET fd, const TPart* parts, size_t count) override { ssize_t ret = SendVImpl(fd, parts, count); if (ret < 0) { return ret; } size_t len = TContIOVector::Bytes(parts, count); if ((size_t)ret == len) { return ret; } return SendVPartial(fd, parts, count, ret); } inline ssize_t SendVImpl(SOCKET fd, const TPart* parts, size_t count) { return TSender < (sizeof(iovec) == sizeof(TPart)) && (offsetof(iovec, iov_base) == offsetof(TPart, buf)) && (offsetof(iovec, iov_len) == offsetof(TPart, len)) > ::SendV(fd, parts, count); } ssize_t SendVPartial(SOCKET fd, const TPart* constParts, size_t count, size_t written); }; ssize_t TCommonSockOps::SendVPartial(SOCKET fd, const TPart* constParts, size_t count, size_t written) { TTempBuf tempbuf(sizeof(TPart) * count); TPart* parts = (TPart*)tempbuf.Data(); for (size_t i = 0; i < count; ++i) { parts[i] = constParts[i]; } TContIOVector vec(parts, count); vec.Proceed(written); while (!vec.Complete()) { ssize_t ret = SendVImpl(fd, vec.Parts(), vec.Count()); if (ret < 0) { return ret; } written += ret; vec.Proceed((size_t)ret); } return written; } static inline TSocket::TOps* GetCommonSockOps() noexcept { return Singleton(); } TSocket::TSocket() : Impl_(new TImpl(INVALID_SOCKET, GetCommonSockOps())) { } TSocket::TSocket(SOCKET fd) : Impl_(new TImpl(fd, GetCommonSockOps())) { } TSocket::TSocket(SOCKET fd, TOps* ops) : Impl_(new TImpl(fd, ops)) { } TSocket::TSocket(const TNetworkAddress& addr) : Impl_(new TImpl(DoConnect(addr.Info(), TInstant::Max()), GetCommonSockOps())) { } TSocket::TSocket(const TNetworkAddress& addr, const TDuration& timeOut) : Impl_(new TImpl(DoConnect(addr.Info(), timeOut.ToDeadLine()), GetCommonSockOps())) { } TSocket::TSocket(const TNetworkAddress& addr, const TInstant& deadLine) : Impl_(new TImpl(DoConnect(addr.Info(), deadLine), GetCommonSockOps())) { } TSocket::~TSocket() = default; SOCKET TSocket::Fd() const noexcept { return Impl_->Fd(); } ssize_t TSocket::Send(const void* data, size_t len) { return Impl_->Send(data, len); } ssize_t TSocket::Recv(void* buf, size_t len) { return Impl_->Recv(buf, len); } ssize_t TSocket::SendV(const TPart* parts, size_t count) { return Impl_->SendV(parts, count); } void TSocket::Close() { Impl_->Close(); } TSocketInput::TSocketInput(const TSocket& s) noexcept : S_(s) { } TSocketInput::~TSocketInput() = default; size_t TSocketInput::DoRead(void* buf, size_t len) { const ssize_t ret = S_.Recv(buf, len); if (ret >= 0) { return (size_t)ret; } ythrow TSystemError(-(int)ret) << "can not read from socket input stream"; } TSocketOutput::TSocketOutput(const TSocket& s) noexcept : S_(s) { } TSocketOutput::~TSocketOutput() { try { Finish(); } catch (...) { // ¯\_(ツ)_/¯ } } void TSocketOutput::DoWrite(const void* buf, size_t len) { size_t send = 0; while (len) { const ssize_t ret = S_.Send(buf, len); if (ret < 0) { ythrow TSystemError(-(int)ret) << "can not write to socket output stream; " << send << " bytes already send"; } buf = (const char*)buf + ret; len -= ret; send += ret; } } void TSocketOutput::DoWriteV(const TPart* parts, size_t count) { const ssize_t ret = S_.SendV(parts, count); if (ret < 0) { ythrow TSystemError(-(int)ret) << "can not writev to socket output stream"; } /* * todo for nonblocking sockets? */ } namespace { // https://bugzilla.mozilla.org/attachment.cgi?id=503263&action=diff struct TLocalNames: public THashSet { inline TLocalNames() { insert("localhost"); insert("localhost.localdomain"); insert("localhost6"); insert("localhost6.localdomain6"); insert("::1"); } inline bool IsLocalName(const char* name) const noexcept { struct sockaddr_in sa; memset(&sa, 0, sizeof(sa)); if (inet_pton(AF_INET, name, &(sa.sin_addr)) == 1) { return (InetToHost(sa.sin_addr.s_addr) >> 24) == 127; } return contains(name); } }; } // namespace class TNetworkAddress::TImpl: public TAtomicRefCount { private: class TAddrInfoDeleter { public: TAddrInfoDeleter(bool useFreeAddrInfo = true) : UseFreeAddrInfo_(useFreeAddrInfo) { } void operator()(struct addrinfo* ai) noexcept { if (!UseFreeAddrInfo_ && ai != NULL) { if (ai->ai_addr != NULL) { free(ai->ai_addr); } struct addrinfo* p; while (ai != NULL) { p = ai; ai = ai->ai_next; free(p->ai_canonname); free(p); } } else if (ai != NULL) { freeaddrinfo(ai); } } private: bool UseFreeAddrInfo_ = true; }; public: inline TImpl(const char* host, ui16 port, int flags) : Info_(nullptr, TAddrInfoDeleter{}) { const TString port_st(ToString(port)); struct addrinfo hints; memset(&hints, 0, sizeof(hints)); hints.ai_flags = flags; hints.ai_family = PF_UNSPEC; hints.ai_socktype = SOCK_STREAM; if (!host) { hints.ai_flags |= AI_PASSIVE; } else { if (!Singleton()->IsLocalName(host)) { hints.ai_flags |= AI_ADDRCONFIG; } } struct addrinfo* pai = NULL; const int error = getaddrinfo(host, port_st.data(), &hints, &pai); if (error) { TAddrInfoDeleter()(pai); ythrow TNetworkResolutionError(error) << ": can not resolve " << host << ":" << port; } Info_.reset(pai); } inline TImpl(const char* path, int flags) : Info_(nullptr, TAddrInfoDeleter{/* useFreeAddrInfo = */ false}) { THolder sockAddr( reinterpret_cast(malloc(sizeof(struct sockaddr_un)))); Y_ENSURE(strlen(path) < sizeof(sockAddr->sun_path), "Unix socket path more than " << sizeof(sockAddr->sun_path)); sockAddr->sun_family = AF_UNIX; strcpy(sockAddr->sun_path, path); TAddrInfoPtr hints(reinterpret_cast(malloc(sizeof(struct addrinfo))), TAddrInfoDeleter{/* useFreeAddrInfo = */ false}); memset(hints.get(), 0, sizeof(*hints)); hints->ai_flags = flags; hints->ai_family = AF_UNIX; hints->ai_socktype = SOCK_STREAM; hints->ai_addrlen = sizeof(*sockAddr); hints->ai_addr = (struct sockaddr*)sockAddr.Release(); Info_.reset(hints.release()); } inline struct addrinfo* Info() const noexcept { return Info_.get(); } private: using TAddrInfoPtr = std::unique_ptr; TAddrInfoPtr Info_; }; TNetworkAddress::TNetworkAddress(const TUnixSocketPath& unixSocketPath, int flags) : Impl_(new TImpl(unixSocketPath.Path.data(), flags)) { } TNetworkAddress::TNetworkAddress(const TString& host, ui16 port, int flags) : Impl_(new TImpl(host.data(), port, flags)) { } TNetworkAddress::TNetworkAddress(const TString& host, ui16 port) : Impl_(new TImpl(host.data(), port, 0)) { } TNetworkAddress::TNetworkAddress(ui16 port) : Impl_(new TImpl(nullptr, port, 0)) { } TNetworkAddress::~TNetworkAddress() = default; struct addrinfo* TNetworkAddress::Info() const noexcept { return Impl_->Info(); } TNetworkResolutionError::TNetworkResolutionError(int error) { const char* errMsg = nullptr; #ifdef _win_ errMsg = LastSystemErrorText(error); // gai_strerror is not thread-safe on Windows #else errMsg = gai_strerror(error); #endif (*this) << errMsg << "(" << error; #if defined(_unix_) if (error == EAI_SYSTEM) { (*this) << "; errno=" << LastSystemError(); } #endif (*this) << "): "; } #if defined(_unix_) static inline int GetFlags(int fd) { const int ret = fcntl(fd, F_GETFL); if (ret == -1) { ythrow TSystemError() << "can not get fd flags"; } return ret; } static inline void SetFlags(int fd, int flags) { if (fcntl(fd, F_SETFL, flags) == -1) { ythrow TSystemError() << "can not set fd flags"; } } static inline void EnableFlag(int fd, int flag) { const int oldf = GetFlags(fd); const int newf = oldf | flag; if (oldf != newf) { SetFlags(fd, newf); } } static inline void DisableFlag(int fd, int flag) { const int oldf = GetFlags(fd); const int newf = oldf & (~flag); if (oldf != newf) { SetFlags(fd, newf); } } static inline void SetFlag(int fd, int flag, bool value) { if (value) { EnableFlag(fd, flag); } else { DisableFlag(fd, flag); } } static inline bool FlagsAreEnabled(int fd, int flags) { return GetFlags(fd) & flags; } #endif #if defined(_win_) static inline void SetNonBlockSocket(SOCKET fd, int value) { unsigned long inbuf = value; unsigned long outbuf = 0; DWORD written = 0; if (!inbuf) { WSAEventSelect(fd, nullptr, 0); } if (WSAIoctl(fd, FIONBIO, &inbuf, sizeof(inbuf), &outbuf, sizeof(outbuf), &written, 0, 0) == SOCKET_ERROR) { ythrow TSystemError() << "can not set non block socket state"; } } static inline bool IsNonBlockSocket(SOCKET fd) { unsigned long buf = 0; if (WSAIoctl(fd, FIONBIO, 0, 0, &buf, sizeof(buf), 0, 0, 0) == SOCKET_ERROR) { ythrow TSystemError() << "can not get non block socket state"; } return buf; } #endif void SetNonBlock(SOCKET fd, bool value) { #if defined(_unix_) #if defined(FIONBIO) Y_UNUSED(SetFlag); // shut up clang about unused function int nb = value; if (ioctl(fd, FIONBIO, &nb) < 0) { ythrow TSystemError() << "ioctl failed"; } #else SetFlag(fd, O_NONBLOCK, value); #endif #elif defined(_win_) SetNonBlockSocket(fd, value); #else #error todo #endif } bool IsNonBlock(SOCKET fd) { #if defined(_unix_) return FlagsAreEnabled(fd, O_NONBLOCK); #elif defined(_win_) return IsNonBlockSocket(fd); #else #error todo #endif } void SetDeferAccept(SOCKET s) { (void)s; #if defined(TCP_DEFER_ACCEPT) CheckedSetSockOpt(s, IPPROTO_TCP, TCP_DEFER_ACCEPT, 10, "defer accept"); #endif #if defined(SO_ACCEPTFILTER) struct accept_filter_arg afa; Zero(afa); strcpy(afa.af_name, "dataready"); SetSockOpt(s, SOL_SOCKET, SO_ACCEPTFILTER, afa); #endif } ssize_t PollD(struct pollfd fds[], nfds_t nfds, const TInstant& deadLine) noexcept { TInstant now = TInstant::Now(); do { const TDuration toWait = PollStep(deadLine, now); const int res = poll(fds, nfds, MicroToMilli(toWait.MicroSeconds())); if (res > 0) { return res; } if (res < 0) { const int err = LastSystemError(); if (err != ETIMEDOUT && err != EINTR) { return -err; } } } while ((now = TInstant::Now()) < deadLine); return -ETIMEDOUT; } void ShutDown(SOCKET s, int mode) { if (shutdown(s, mode)) { ythrow TSystemError() << "shutdown socket error"; } }