1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932
|
// Copyright (c) Meta Platforms, Inc. and its affiliates.
// All rights reserved.
//
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree.
#include <torch/csrc/distributed/c10d/socket.h>
#include <cstring>
#include <system_error>
#include <thread>
#include <utility>
#include <vector>
#ifdef _WIN32
#include <mutex>
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <fcntl.h>
#include <netdb.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#endif
#include <fmt/chrono.h>
#include <fmt/format.h>
#include <torch/csrc/distributed/c10d/error.h>
#include <torch/csrc/distributed/c10d/exception.h>
#include <torch/csrc/distributed/c10d/logging.h>
#include <c10/util/CallOnce.h>
namespace c10d {
namespace detail {
namespace {
#ifdef _WIN32
// Since Winsock uses the name `WSAPoll` instead of `poll`, we alias it here
// to avoid #ifdefs in the source code.
const auto pollFd = ::WSAPoll;
// Winsock's `getsockopt()` and `setsockopt()` functions expect option values to
// be passed as `char*` instead of `void*`. We wrap them here to avoid redundant
// casts in the source code.
int getSocketOption(
SOCKET s,
int level,
int optname,
void* optval,
int* optlen) {
return ::getsockopt(s, level, optname, static_cast<char*>(optval), optlen);
}
int setSocketOption(
SOCKET s,
int level,
int optname,
const void* optval,
int optlen) {
return ::setsockopt(
s, level, optname, static_cast<const char*>(optval), optlen);
}
// Winsock has its own error codes which differ from Berkeley's. Fortunately the
// C++ Standard Library on Windows can map them to standard error codes.
inline std::error_code getSocketError() noexcept {
return std::error_code{::WSAGetLastError(), std::system_category()};
}
inline void setSocketError(int val) noexcept {
::WSASetLastError(val);
}
#else
const auto pollFd = ::poll;
const auto getSocketOption = ::getsockopt;
const auto setSocketOption = ::setsockopt;
inline std::error_code getSocketError() noexcept {
return lastError();
}
inline void setSocketError(int val) noexcept {
errno = val;
}
#endif
// Suspends the current thread for the specified duration.
void delay(std::chrono::seconds d) {
#ifdef _WIN32
std::this_thread::sleep_for(d);
#else
::timespec req{};
req.tv_sec = d.count();
// The C++ Standard does not specify whether `sleep_for()` should be signal-
// aware; therefore, we use the `nanosleep()` syscall.
if (::nanosleep(&req, nullptr) != 0) {
std::error_code err = getSocketError();
// We don't care about error conditions other than EINTR since a failure
// here is not critical.
if (err == std::errc::interrupted) {
throw std::system_error{err};
}
}
#endif
}
class SocketListenOp;
class SocketConnectOp;
} // namespace
class SocketImpl {
friend class SocketListenOp;
friend class SocketConnectOp;
public:
#ifdef _WIN32
using Handle = SOCKET;
#else
using Handle = int;
#endif
#ifdef _WIN32
static constexpr Handle invalid_socket = INVALID_SOCKET;
#else
static constexpr Handle invalid_socket = -1;
#endif
explicit SocketImpl(Handle hnd) noexcept : hnd_{hnd} {}
SocketImpl(const SocketImpl& other) = delete;
SocketImpl& operator=(const SocketImpl& other) = delete;
SocketImpl(SocketImpl&& other) noexcept = delete;
SocketImpl& operator=(SocketImpl&& other) noexcept = delete;
~SocketImpl();
std::unique_ptr<SocketImpl> accept() const;
void closeOnExec() noexcept;
void enableNonBlocking();
void disableNonBlocking();
bool enableNoDelay() noexcept;
bool enableDualStack() noexcept;
#ifndef _WIN32
bool enableAddressReuse() noexcept;
#endif
#ifdef _WIN32
bool enableExclusiveAddressUse() noexcept;
#endif
std::uint16_t getPort() const;
Handle handle() const noexcept {
return hnd_;
}
private:
bool setSocketFlag(int level, int optname, bool value) noexcept;
Handle hnd_;
};
} // namespace detail
} // namespace c10d
//
// libfmt formatters for `addrinfo` and `Socket`
//
namespace fmt {
template <>
struct formatter<::addrinfo> {
constexpr decltype(auto) parse(format_parse_context& ctx) {
return ctx.begin();
}
template <typename FormatContext>
decltype(auto) format(const ::addrinfo& addr, FormatContext& ctx) {
char host[NI_MAXHOST], port[NI_MAXSERV]; // NOLINT
int r = ::getnameinfo(
addr.ai_addr,
addr.ai_addrlen,
host,
NI_MAXHOST,
port,
NI_MAXSERV,
NI_NUMERICSERV);
if (r != 0) {
return format_to(ctx.out(), "?UNKNOWN?");
}
if (addr.ai_addr->sa_family == AF_INET) {
return format_to(ctx.out(), "{}:{}", host, port);
} else {
return format_to(ctx.out(), "[{}]:{}", host, port);
}
}
};
template <>
struct formatter<c10d::detail::SocketImpl> {
constexpr decltype(auto) parse(format_parse_context& ctx) {
return ctx.begin();
}
template <typename FormatContext>
decltype(auto) format(
const c10d::detail::SocketImpl& socket,
FormatContext& ctx) {
::sockaddr_storage addr_s{};
auto addr_ptr = reinterpret_cast<::sockaddr*>(&addr_s);
::socklen_t addr_len = sizeof(addr_s);
if (::getsockname(socket.handle(), addr_ptr, &addr_len) != 0) {
return format_to(ctx.out(), "?UNKNOWN?");
}
::addrinfo addr{};
addr.ai_addr = addr_ptr;
addr.ai_addrlen = addr_len;
return format_to(ctx.out(), "{}", addr);
}
};
} // namespace fmt
namespace c10d {
namespace detail {
SocketImpl::~SocketImpl() {
#ifdef _WIN32
::closesocket(hnd_);
#else
::close(hnd_);
#endif
}
std::unique_ptr<SocketImpl> SocketImpl::accept() const {
::sockaddr_storage addr_s{};
auto addr_ptr = reinterpret_cast<::sockaddr*>(&addr_s);
::socklen_t addr_len = sizeof(addr_s);
Handle hnd = ::accept(hnd_, addr_ptr, &addr_len);
if (hnd == invalid_socket) {
std::error_code err = getSocketError();
if (err == std::errc::interrupted) {
throw std::system_error{err};
}
std::string msg{};
if (err == std::errc::invalid_argument) {
msg = fmt::format(
"The server socket on {} is not listening for connections.", *this);
} else {
msg = fmt::format(
"The server socket on {} has failed to accept a connection {}.",
*this,
err);
}
C10D_ERROR(msg);
throw SocketError{msg};
}
::addrinfo addr{};
addr.ai_addr = addr_ptr;
addr.ai_addrlen = addr_len;
C10D_DEBUG(
"The server socket on {} has accepted a connection from {}.",
*this,
addr);
auto impl = std::make_unique<SocketImpl>(hnd);
// Make sure that we do not "leak" our file descriptors to child processes.
impl->closeOnExec();
if (!impl->enableNoDelay()) {
C10D_WARNING(
"The no-delay option cannot be enabled for the client socket on {}.",
addr);
}
return impl;
}
void SocketImpl::closeOnExec() noexcept {
#ifndef _WIN32
::fcntl(hnd_, F_SETFD, FD_CLOEXEC);
#endif
}
void SocketImpl::enableNonBlocking() {
#ifdef _WIN32
unsigned long value = 1;
if (::ioctlsocket(hnd_, FIONBIO, &value) == 0) {
return;
}
#else
int flg = ::fcntl(hnd_, F_GETFL);
if (flg != -1) {
if (::fcntl(hnd_, F_SETFL, flg | O_NONBLOCK) == 0) {
return;
}
}
#endif
throw SocketError{"The socket cannot be switched to non-blocking mode."};
}
// TODO: Remove once we migrate everything to non-blocking mode.
void SocketImpl::disableNonBlocking() {
#ifdef _WIN32
unsigned long value = 0;
if (::ioctlsocket(hnd_, FIONBIO, &value) == 0) {
return;
}
#else
int flg = ::fcntl(hnd_, F_GETFL);
if (flg != -1) {
if (::fcntl(hnd_, F_SETFL, flg & ~O_NONBLOCK) == 0) {
return;
}
}
#endif
throw SocketError{"The socket cannot be switched to blocking mode."};
}
bool SocketImpl::enableNoDelay() noexcept {
return setSocketFlag(IPPROTO_TCP, TCP_NODELAY, true);
}
bool SocketImpl::enableDualStack() noexcept {
return setSocketFlag(IPPROTO_IPV6, IPV6_V6ONLY, false);
}
#ifndef _WIN32
bool SocketImpl::enableAddressReuse() noexcept {
return setSocketFlag(SOL_SOCKET, SO_REUSEADDR, true);
}
#endif
#ifdef _WIN32
bool SocketImpl::enableExclusiveAddressUse() noexcept {
return setSocketFlag(SOL_SOCKET, SO_EXCLUSIVEADDRUSE, true);
}
#endif
std::uint16_t SocketImpl::getPort() const {
::sockaddr_storage addr_s{};
::socklen_t addr_len = sizeof(addr_s);
if (::getsockname(hnd_, reinterpret_cast<::sockaddr*>(&addr_s), &addr_len) !=
0) {
throw SocketError{"The port number of the socket cannot be retrieved."};
}
if (addr_s.ss_family == AF_INET) {
return ntohs(reinterpret_cast<::sockaddr_in*>(&addr_s)->sin_port);
} else {
return ntohs(reinterpret_cast<::sockaddr_in6*>(&addr_s)->sin6_port);
}
}
bool SocketImpl::setSocketFlag(int level, int optname, bool value) noexcept {
#ifdef _WIN32
auto buf = value ? TRUE : FALSE;
#else
auto buf = value ? 1 : 0;
#endif
return setSocketOption(hnd_, level, optname, &buf, sizeof(buf)) == 0;
}
namespace {
struct addrinfo_delete {
void operator()(::addrinfo* addr) const noexcept {
::freeaddrinfo(addr);
}
};
using addrinfo_ptr = std::unique_ptr<::addrinfo, addrinfo_delete>;
class SocketListenOp {
public:
SocketListenOp(std::uint16_t port, const SocketOptions& opts);
std::unique_ptr<SocketImpl> run();
private:
bool tryListen(int family);
bool tryListen(const ::addrinfo& addr);
template <typename... Args>
void recordError(fmt::string_view format, Args&&... args) {
auto msg = fmt::vformat(format, fmt::make_format_args(args...));
C10D_WARNING(msg);
errors_.emplace_back(std::move(msg));
}
std::string port_;
const SocketOptions* opts_;
std::vector<std::string> errors_{};
std::unique_ptr<SocketImpl> socket_{};
};
SocketListenOp::SocketListenOp(std::uint16_t port, const SocketOptions& opts)
: port_{fmt::to_string(port)}, opts_{&opts} {}
std::unique_ptr<SocketImpl> SocketListenOp::run() {
if (opts_->prefer_ipv6()) {
C10D_DEBUG("The server socket will attempt to listen on an IPv6 address.");
if (tryListen(AF_INET6)) {
return std::move(socket_);
}
C10D_DEBUG("The server socket will attempt to listen on an IPv4 address.");
if (tryListen(AF_INET)) {
return std::move(socket_);
}
} else {
C10D_DEBUG(
"The server socket will attempt to listen on an IPv4 or IPv6 address.");
if (tryListen(AF_UNSPEC)) {
return std::move(socket_);
}
}
constexpr auto* msg =
"The server socket has failed to listen on any local network address.";
C10D_ERROR(msg);
throw SocketError{fmt::format("{} {}", msg, fmt::join(errors_, " "))};
}
bool SocketListenOp::tryListen(int family) {
::addrinfo hints{}, *naked_result = nullptr;
hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
hints.ai_family = family;
hints.ai_socktype = SOCK_STREAM;
int r = ::getaddrinfo(nullptr, port_.c_str(), &hints, &naked_result);
if (r != 0) {
const char* gai_err = ::gai_strerror(r);
recordError(
"The local {}network addresses cannot be retrieved (gai error: {} - {}).",
family == AF_INET ? "IPv4 "
: family == AF_INET6 ? "IPv6 "
: "",
r,
gai_err);
return false;
}
addrinfo_ptr result{naked_result};
for (::addrinfo* addr = naked_result; addr != nullptr; addr = addr->ai_next) {
C10D_DEBUG("The server socket is attempting to listen on {}.", *addr);
if (tryListen(*addr)) {
return true;
}
}
return false;
}
bool SocketListenOp::tryListen(const ::addrinfo& addr) {
SocketImpl::Handle hnd =
::socket(addr.ai_family, addr.ai_socktype, addr.ai_protocol);
if (hnd == SocketImpl::invalid_socket) {
recordError(
"The server socket cannot be initialized on {} {}.",
addr,
getSocketError());
return false;
}
socket_ = std::make_unique<SocketImpl>(hnd);
#ifndef _WIN32
if (!socket_->enableAddressReuse()) {
C10D_WARNING(
"The address reuse option cannot be enabled for the server socket on {}.",
addr);
}
#endif
#ifdef _WIN32
// The SO_REUSEADDR flag has a significantly different behavior on Windows
// compared to Unix-like systems. It allows two or more processes to share
// the same port simultaneously, which is totally unsafe.
//
// Here we follow the recommendation of Microsoft and use the non-standard
// SO_EXCLUSIVEADDRUSE flag instead.
if (!socket_->enableExclusiveAddressUse()) {
C10D_WARNING(
"The exclusive address use option cannot be enabled for the server socket on {}.",
addr);
}
#endif
// Not all operating systems support dual-stack sockets by default. Since we
// wish to use our IPv6 socket for IPv4 communication as well, we explicitly
// ask the system to enable it.
if (addr.ai_family == AF_INET6 && !socket_->enableDualStack()) {
C10D_WARNING(
"The server socket does not support IPv4 communication on {}.", addr);
}
if (::bind(socket_->handle(), addr.ai_addr, addr.ai_addrlen) != 0) {
recordError(
"The server socket has failed to bind to {} {}.",
addr,
getSocketError());
return false;
}
// NOLINTNEXTLINE(bugprone-argument-comment)
if (::listen(socket_->handle(), /*backlog=*/2048) != 0) {
recordError(
"The server socket has failed to listen on {} {}.",
addr,
getSocketError());
return false;
}
socket_->closeOnExec();
C10D_INFO("The server socket has started to listen on {}.", addr);
return true;
}
class SocketConnectOp {
using Clock = std::chrono::steady_clock;
using Duration = std::chrono::steady_clock::duration;
using TimePoint = std::chrono::time_point<std::chrono::steady_clock>;
static const std::chrono::seconds delay_duration_;
enum class ConnectResult { Success, Error, Retry };
public:
SocketConnectOp(
const std::string& host,
std::uint16_t port,
const SocketOptions& opts);
std::unique_ptr<SocketImpl> run();
private:
bool tryConnect(int family);
ConnectResult tryConnect(const ::addrinfo& addr);
ConnectResult tryConnectCore(const ::addrinfo& addr);
[[noreturn]] void throwTimeoutError() const;
template <typename... Args>
void recordError(fmt::string_view format, Args&&... args) {
auto msg = fmt::vformat(format, fmt::make_format_args(args...));
C10D_WARNING(msg);
errors_.emplace_back(std::move(msg));
}
const char* host_;
std::string port_;
const SocketOptions* opts_;
TimePoint deadline_{};
std::vector<std::string> errors_{};
std::unique_ptr<SocketImpl> socket_{};
};
const std::chrono::seconds SocketConnectOp::delay_duration_{1};
SocketConnectOp::SocketConnectOp(
const std::string& host,
std::uint16_t port,
const SocketOptions& opts)
: host_{host.c_str()}, port_{fmt::to_string(port)}, opts_{&opts} {}
std::unique_ptr<SocketImpl> SocketConnectOp::run() {
if (opts_->prefer_ipv6()) {
C10D_DEBUG(
"The client socket will attempt to connect to an IPv6 address of ({}, {}).",
host_,
port_);
if (tryConnect(AF_INET6)) {
return std::move(socket_);
}
C10D_DEBUG(
"The client socket will attempt to connect to an IPv4 address of ({}, {}).",
host_,
port_);
if (tryConnect(AF_INET)) {
return std::move(socket_);
}
} else {
C10D_DEBUG(
"The client socket will attempt to connect to an IPv4 or IPv6 address of ({}, {}).",
host_,
port_);
if (tryConnect(AF_UNSPEC)) {
return std::move(socket_);
}
}
auto msg = fmt::format(
"The client socket has failed to connect to any network address of ({}, {}).",
host_,
port_);
C10D_ERROR(msg);
throw SocketError{fmt::format("{} {}", msg, fmt::join(errors_, " "))};
}
bool SocketConnectOp::tryConnect(int family) {
::addrinfo hints{};
hints.ai_flags = AI_V4MAPPED | AI_ALL | AI_NUMERICSERV;
hints.ai_family = family;
hints.ai_socktype = SOCK_STREAM;
deadline_ = Clock::now() + opts_->connect_timeout();
std::size_t retry_attempt = 1;
bool retry; // NOLINT(cppcoreguidelines-init-variables)
do {
retry = false;
errors_.clear();
::addrinfo* naked_result = nullptr;
// patternlint-disable cpp-dns-deps
int r = ::getaddrinfo(host_, port_.c_str(), &hints, &naked_result);
if (r != 0) {
const char* gai_err = ::gai_strerror(r);
recordError(
"The {}network addresses of ({}, {}) cannot be retrieved (gai error: {} - {}).",
family == AF_INET ? "IPv4 "
: family == AF_INET6 ? "IPv6 "
: "",
host_,
port_,
r,
gai_err);
retry = true;
} else {
addrinfo_ptr result{naked_result};
for (::addrinfo* addr = naked_result; addr != nullptr;
addr = addr->ai_next) {
C10D_TRACE("The client socket is attempting to connect to {}.", *addr);
ConnectResult cr = tryConnect(*addr);
if (cr == ConnectResult::Success) {
return true;
}
if (cr == ConnectResult::Retry) {
retry = true;
}
}
}
if (retry) {
if (Clock::now() < deadline_ - delay_duration_) {
// Prevent our log output to be too noisy, warn only every 30 seconds.
if (retry_attempt == 30) {
C10D_INFO(
"No socket on ({}, {}) is listening yet, will retry.",
host_,
port_);
retry_attempt = 0;
}
// Wait one second to avoid choking the server.
delay(delay_duration_);
retry_attempt++;
} else {
throwTimeoutError();
}
}
} while (retry);
return false;
}
SocketConnectOp::ConnectResult SocketConnectOp::tryConnect(
const ::addrinfo& addr) {
if (Clock::now() >= deadline_) {
throwTimeoutError();
}
SocketImpl::Handle hnd =
::socket(addr.ai_family, addr.ai_socktype, addr.ai_protocol);
if (hnd == SocketImpl::invalid_socket) {
recordError(
"The client socket cannot be initialized to connect to {} {}.",
addr,
getSocketError());
return ConnectResult::Error;
}
socket_ = std::make_unique<SocketImpl>(hnd);
socket_->enableNonBlocking();
ConnectResult cr = tryConnectCore(addr);
if (cr == ConnectResult::Error) {
std::error_code err = getSocketError();
if (err == std::errc::interrupted) {
throw std::system_error{err};
}
// Retry if the server is not yet listening or if its backlog is exhausted.
if (err == std::errc::connection_refused ||
err == std::errc::connection_reset) {
C10D_TRACE(
"The server socket on {} is not yet listening {}, will retry.",
addr,
err);
return ConnectResult::Retry;
} else {
recordError(
"The client socket has failed to connect to {} {}.", addr, err);
return ConnectResult::Error;
}
}
socket_->closeOnExec();
// TODO: Remove once we fully migrate to non-blocking mode.
socket_->disableNonBlocking();
C10D_INFO("The client socket has connected to {} on {}.", addr, *socket_);
if (!socket_->enableNoDelay()) {
C10D_WARNING(
"The no-delay option cannot be enabled for the client socket on {}.",
*socket_);
}
return ConnectResult::Success;
}
SocketConnectOp::ConnectResult SocketConnectOp::tryConnectCore(
const ::addrinfo& addr) {
int r = ::connect(socket_->handle(), addr.ai_addr, addr.ai_addrlen);
if (r == 0) {
return ConnectResult::Success;
}
std::error_code err = getSocketError();
if (err == std::errc::already_connected) {
return ConnectResult::Success;
}
if (err != std::errc::operation_in_progress &&
err != std::errc::operation_would_block) {
return ConnectResult::Error;
}
Duration remaining = deadline_ - Clock::now();
if (remaining <= Duration::zero()) {
throwTimeoutError();
}
::pollfd pfd{};
pfd.fd = socket_->handle();
pfd.events = POLLOUT;
auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(remaining);
r = pollFd(&pfd, 1, static_cast<int>(ms.count()));
if (r == 0) {
throwTimeoutError();
}
if (r == -1) {
return ConnectResult::Error;
}
int err_code = 0;
::socklen_t err_len = sizeof(int);
r = getSocketOption(
socket_->handle(), SOL_SOCKET, SO_ERROR, &err_code, &err_len);
if (r != 0) {
return ConnectResult::Error;
}
if (err_code != 0) {
setSocketError(err_code);
return ConnectResult::Error;
} else {
return ConnectResult::Success;
}
}
void SocketConnectOp::throwTimeoutError() const {
auto msg = fmt::format(
"The client socket has timed out after {} while trying to connect to ({}, {}).",
opts_->connect_timeout(),
host_,
port_);
C10D_ERROR(msg);
throw TimeoutError{msg};
}
} // namespace
void Socket::initialize() {
#ifdef _WIN32
static c10::once_flag init_flag{};
// All processes that call socket functions on Windows must first initialize
// the Winsock library.
c10::call_once(init_flag, []() {
WSADATA data{};
if (::WSAStartup(MAKEWORD(2, 2), &data) != 0) {
throw SocketError{"The initialization of Winsock has failed."};
}
});
#endif
}
Socket Socket::listen(std::uint16_t port, const SocketOptions& opts) {
SocketListenOp op{port, opts};
return Socket{op.run()};
}
Socket Socket::connect(
const std::string& host,
std::uint16_t port,
const SocketOptions& opts) {
SocketConnectOp op{host, port, opts};
return Socket{op.run()};
}
Socket::Socket(Socket&& other) noexcept = default;
Socket& Socket::operator=(Socket&& other) noexcept = default;
Socket::~Socket() = default;
Socket Socket::accept() const {
if (impl_) {
return Socket{impl_->accept()};
}
throw SocketError{"The socket is not initialized."};
}
int Socket::handle() const noexcept {
if (impl_) {
return impl_->handle();
}
return SocketImpl::invalid_socket;
}
std::uint16_t Socket::port() const {
if (impl_) {
return impl_->getPort();
}
return 0;
}
Socket::Socket(std::unique_ptr<SocketImpl>&& impl) noexcept
: impl_{std::move(impl)} {}
} // namespace detail
SocketError::~SocketError() = default;
} // namespace c10d
|