folly/io/async/AsyncServerSocket.cpp

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #ifndef __STDC_FORMAT_MACROS
  18   #define __STDC_FORMAT_MACROS
  19 #endif
  20
  21 #include <folly/io/async/AsyncServerSocket.h>
  22
  23 #include <folly/FileUtil.h>
  24 #include <folly/Portability.h>
  25 #include <folly/SocketAddress.h>
  26 #include <folly/String.h>
  27 #include <folly/detail/SocketFastOpen.h>
  28 #include <folly/io/async/EventBase.h>
  29 #include <folly/io/async/NotificationQueue.h>
  30 #include <folly/portability/Fcntl.h>
  31 #include <folly/portability/Sockets.h>
  32 #include <folly/portability/Unistd.h>
  33
  34 #include <errno.h>
  35 #include <string.h>
  36 #include <sys/types.h>
  37
  38 namespace fsp = folly::portability::sockets;
  39
  40 namespace folly {
  41
  42 static constexpr bool msgErrQueueSupported =
  43 #ifdef MSG_ERRQUEUE
  44     true;
  45 #else
  46     false;
  47 #endif // MSG_ERRQUEUE
  48
  49 const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce;
  50 const uint32_t AsyncServerSocket::kDefaultCallbackAcceptAtOnce;
  51 const uint32_t AsyncServerSocket::kDefaultMaxMessagesInQueue;
  52
  53 int setCloseOnExec(int fd, int value) {
  54   // Read the current flags
  55   int old_flags = fcntl(fd, F_GETFD, 0);
  56
  57   // If reading the flags failed, return error indication now
  58   if (old_flags < 0)
  59     return -1;
  60
  61   // Set just the flag we want to set
  62   int new_flags;
  63   if (value != 0)
  64     new_flags = old_flags | FD_CLOEXEC;
  65   else
  66     new_flags = old_flags & ~FD_CLOEXEC;
  67
  68   // Store modified flag word in the descriptor
  69   return fcntl(fd, F_SETFD, new_flags);
  70 }
  71
  72 void AsyncServerSocket::RemoteAcceptor::start(
  73   EventBase* eventBase, uint32_t maxAtOnce, uint32_t maxInQueue) {
  74   setMaxReadAtOnce(maxAtOnce);
  75   queue_.setMaxQueueSize(maxInQueue);
  76
  77   if (!eventBase->runInEventBaseThread([=](){
  78         callback_->acceptStarted();
  79         this->startConsuming(eventBase, &queue_);
  80       })) {
  81     throw std::invalid_argument("unable to start waiting on accept "
  82                             "notification queue in the specified "
  83                             "EventBase thread");
  84   }
  85 }
  86
  87 void AsyncServerSocket::RemoteAcceptor::stop(
  88   EventBase* eventBase, AcceptCallback* callback) {
  89   if (!eventBase->runInEventBaseThread([=](){
  90         callback->acceptStopped();
  91         delete this;
  92       })) {
  93     throw std::invalid_argument("unable to start waiting on accept "
  94                             "notification queue in the specified "
  95                             "EventBase thread");
  96   }
  97 }
  98
  99 void AsyncServerSocket::RemoteAcceptor::messageAvailable(
 100     QueueMessage&& msg) noexcept {
 101   switch (msg.type) {
 102     case MessageType::MSG_NEW_CONN:
 103     {
 104       if (connectionEventCallback_) {
 105         connectionEventCallback_->onConnectionDequeuedByAcceptorCallback(
 106             msg.fd, msg.address);
 107       }
 108       callback_->connectionAccepted(msg.fd, msg.address);
 109       break;
 110     }
 111     case MessageType::MSG_ERROR:
 112     {
 113       std::runtime_error ex(msg.msg);
 114       callback_->acceptError(ex);
 115       break;
 116     }
 117     default:
 118     {
 119       LOG(ERROR) << "invalid accept notification message type "
 120                  << int(msg.type);
 121       std::runtime_error ex(
 122         "received invalid accept notification message type");
 123       callback_->acceptError(ex);
 124     }
 125   }
 126 }
 127
 128 /*
 129  * AsyncServerSocket::BackoffTimeout
 130  */
 131 class AsyncServerSocket::BackoffTimeout : public AsyncTimeout {
 132  public:
 133   // Disallow copy, move, and default constructors.
 134   BackoffTimeout(BackoffTimeout&&) = delete;
 135   explicit BackoffTimeout(AsyncServerSocket* socket)
 136       : AsyncTimeout(socket->getEventBase()), socket_(socket) {}
 137
 138   void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); }
 139
 140  private:
 141   AsyncServerSocket* socket_;
 142 };
 143
 144 /*
 145  * AsyncServerSocket methods
 146  */
 147
 148 AsyncServerSocket::AsyncServerSocket(EventBase* eventBase)
 149 :   eventBase_(eventBase),
 150     accepting_(false),
 151     maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce),
 152     maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue),
 153     acceptRateAdjustSpeed_(0),
 154     acceptRate_(1),
 155     lastAccepTimestamp_(std::chrono::steady_clock::now()),
 156     numDroppedConnections_(0),
 157     callbackIndex_(0),
 158     backoffTimeout_(nullptr),
 159     callbacks_(),
 160     keepAliveEnabled_(true),
 161     closeOnExec_(true),
 162     shutdownSocketSet_(nullptr) {
 163 }
 164
 165 void AsyncServerSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
 166   if (shutdownSocketSet_ == newSS) {
 167     return;
 168   }
 169   if (shutdownSocketSet_) {
 170     for (auto& h : sockets_) {
 171       shutdownSocketSet_->remove(h.socket_);
 172     }
 173   }
 174   shutdownSocketSet_ = newSS;
 175   if (shutdownSocketSet_) {
 176     for (auto& h : sockets_) {
 177       shutdownSocketSet_->add(h.socket_);
 178     }
 179   }
 180 }
 181
 182 AsyncServerSocket::~AsyncServerSocket() {
 183   assert(callbacks_.empty());
 184 }
 185
 186 int AsyncServerSocket::stopAccepting(int shutdownFlags) {
 187   int result = 0;
 188   for (auto& handler : sockets_) {
 189     VLOG(10) << "AsyncServerSocket::stopAccepting " << this <<
 190               handler.socket_;
 191   }
 192   if (eventBase_) {
 193     eventBase_->dcheckIsInEventBaseThread();
 194   }
 195
 196   // When destroy is called, unregister and close the socket immediately.
 197   accepting_ = false;
 198
 199   // Close the sockets in reverse order as they were opened to avoid
 200   // the condition where another process concurrently tries to open
 201   // the same port, succeed to bind the first socket but fails on the
 202   // second because it hasn't been closed yet.
 203   for (; !sockets_.empty(); sockets_.pop_back()) {
 204     auto& handler = sockets_.back();
 205     handler.unregisterHandler();
 206     if (shutdownSocketSet_) {
 207       shutdownSocketSet_->close(handler.socket_);
 208     } else if (shutdownFlags >= 0) {
 209       result = shutdownNoInt(handler.socket_, shutdownFlags);
 210       pendingCloseSockets_.push_back(handler.socket_);
 211     } else {
 212       closeNoInt(handler.socket_);
 213     }
 214   }
 215
 216   // Destroy the backoff timout.  This will cancel it if it is running.
 217   delete backoffTimeout_;
 218   backoffTimeout_ = nullptr;
 219
 220   // Close all of the callback queues to notify them that they are being
 221   // destroyed.  No one should access the AsyncServerSocket any more once
 222   // destroy() is called.  However, clear out callbacks_ before invoking the
 223   // accept callbacks just in case.  This will potentially help us detect the
 224   // bug if one of the callbacks calls addAcceptCallback() or
 225   // removeAcceptCallback().
 226   std::vector<CallbackInfo> callbacksCopy;
 227   callbacks_.swap(callbacksCopy);
 228   for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
 229        it != callbacksCopy.end();
 230        ++it) {
 231     // consumer may not be set if we are running in primary event base
 232     if (it->consumer) {
 233       DCHECK(it->eventBase);
 234       it->consumer->stop(it->eventBase, it->callback);
 235     } else {
 236       DCHECK(it->callback);
 237       it->callback->acceptStopped();
 238     }
 239   }
 240
 241   return result;
 242 }
 243
 244 void AsyncServerSocket::destroy() {
 245   stopAccepting();
 246   for (auto s : pendingCloseSockets_) {
 247     closeNoInt(s);
 248   }
 249   // Then call DelayedDestruction::destroy() to take care of
 250   // whether or not we need immediate or delayed destruction
 251   DelayedDestruction::destroy();
 252 }
 253
 254 void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
 255   assert(eventBase_ == nullptr);
 256   eventBase->dcheckIsInEventBaseThread();
 257
 258   eventBase_ = eventBase;
 259   for (auto& handler : sockets_) {
 260     handler.attachEventBase(eventBase);
 261   }
 262 }
 263
 264 void AsyncServerSocket::detachEventBase() {
 265   assert(eventBase_ != nullptr);
 266   eventBase_->dcheckIsInEventBaseThread();
 267   assert(!accepting_);
 268
 269   eventBase_ = nullptr;
 270   for (auto& handler : sockets_) {
 271     handler.detachEventBase();
 272   }
 273 }
 274
 275 void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
 276   if (eventBase_) {
 277     eventBase_->dcheckIsInEventBaseThread();
 278   }
 279
 280   if (sockets_.size() > 0) {
 281     throw std::invalid_argument(
 282                               "cannot call useExistingSocket() on a "
 283                               "AsyncServerSocket that already has a socket");
 284   }
 285
 286   for (auto fd: fds) {
 287     // Set addressFamily_ from this socket.
 288     // Note that the socket may not have been bound yet, but
 289     // setFromLocalAddress() will still work and get the correct address family.
 290     // We will update addressFamily_ again anyway if bind() is called later.
 291     SocketAddress address;
 292     address.setFromLocalAddress(fd);
 293
 294 #if __linux__
 295     if (noTransparentTls_) {
 296       // Ignore return value, errors are ok
 297       setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
 298     }
 299 #endif
 300
 301     setupSocket(fd, address.getFamily());
 302     sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
 303     sockets_.back().changeHandlerFD(fd);
 304   }
 305 }
 306
 307 void AsyncServerSocket::useExistingSocket(int fd) {
 308   useExistingSockets({fd});
 309 }
 310
 311 void AsyncServerSocket::bindSocket(
 312     int fd,
 313     const SocketAddress& address,
 314     bool isExistingSocket) {
 315   sockaddr_storage addrStorage;
 316   address.getAddress(&addrStorage);
 317   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 318
 319   if (fsp::bind(fd, saddr, address.getActualSize()) != 0) {
 320     if (!isExistingSocket) {
 321       closeNoInt(fd);
 322     }
 323     folly::throwSystemError(errno,
 324         "failed to bind to async server socket: " +
 325         address.describe());
 326   }
 327
 328 #if __linux__
 329   if (noTransparentTls_) {
 330     // Ignore return value, errors are ok
 331     setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
 332   }
 333 #endif
 334
 335   // If we just created this socket, update the EventHandler and set socket_
 336   if (!isExistingSocket) {
 337     sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
 338   }
 339 }
 340
 341 bool AsyncServerSocket::setZeroCopy(bool enable) {
 342   if (msgErrQueueSupported) {
 343     int fd = getSocket();
 344     int val = enable ? 1 : 0;
 345     int ret = setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, &val, sizeof(val));
 346
 347     return (0 == ret);
 348   }
 349
 350   return false;
 351 }
 352
 353 void AsyncServerSocket::bind(const SocketAddress& address) {
 354   if (eventBase_) {
 355     eventBase_->dcheckIsInEventBaseThread();
 356   }
 357
 358   // useExistingSocket() may have been called to initialize socket_ already.
 359   // However, in the normal case we need to create a new socket now.
 360   // Don't set socket_ yet, so that socket_ will remain uninitialized if an
 361   // error occurs.
 362   int fd;
 363   if (sockets_.size() == 0) {
 364     fd = createSocket(address.getFamily());
 365   } else if (sockets_.size() == 1) {
 366     if (address.getFamily() != sockets_[0].addressFamily_) {
 367       throw std::invalid_argument(
 368                                 "Attempted to bind address to socket with "
 369                                 "different address family");
 370     }
 371     fd = sockets_[0].socket_;
 372   } else {
 373     throw std::invalid_argument(
 374                               "Attempted to bind to multiple fds");
 375   }
 376
 377   bindSocket(fd, address, !sockets_.empty());
 378 }
 379
 380 void AsyncServerSocket::bind(
 381     const std::vector<IPAddress>& ipAddresses,
 382     uint16_t port) {
 383   if (ipAddresses.empty()) {
 384     throw std::invalid_argument("No ip addresses were provided");
 385   }
 386   if (!sockets_.empty()) {
 387     throw std::invalid_argument("Cannot call bind on a AsyncServerSocket "
 388                                 "that already has a socket.");
 389   }
 390
 391   for (const IPAddress& ipAddress : ipAddresses) {
 392     SocketAddress address(ipAddress.toFullyQualified(), port);
 393     int fd = createSocket(address.getFamily());
 394
 395     bindSocket(fd, address, false);
 396   }
 397   if (sockets_.size() == 0) {
 398     throw std::runtime_error(
 399         "did not bind any async server socket for port and addresses");
 400   }
 401 }
 402
 403 void AsyncServerSocket::bind(uint16_t port) {
 404   struct addrinfo hints, *res0;
 405   char sport[sizeof("65536")];
 406
 407   memset(&hints, 0, sizeof(hints));
 408   hints.ai_family = AF_UNSPEC;
 409   hints.ai_socktype = SOCK_STREAM;
 410   hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
 411   snprintf(sport, sizeof(sport), "%u", port);
 412
 413   // On Windows the value we need to pass to bind to all available
 414   // addresses is an empty string. Everywhere else, it's nullptr.
 415   constexpr const char* kWildcardNode = kIsWindows ? "" : nullptr;
 416   if (getaddrinfo(kWildcardNode, sport, &hints, &res0)) {
 417     throw std::invalid_argument(
 418                               "Attempted to bind address to socket with "
 419                               "bad getaddrinfo");
 420   }
 421
 422   SCOPE_EXIT { freeaddrinfo(res0); };
 423
 424   auto setupAddress = [&] (struct addrinfo* res) {
 425     int s = fsp::socket(res->ai_family, res->ai_socktype, res->ai_protocol);
 426     // IPv6/IPv4 may not be supported by the kernel
 427     if (s < 0 && errno == EAFNOSUPPORT) {
 428       return;
 429     }
 430     CHECK_GE(s, 0);
 431
 432     try {
 433       setupSocket(s, res->ai_family);
 434     } catch (...) {
 435       closeNoInt(s);
 436       throw;
 437     }
 438
 439     if (res->ai_family == AF_INET6) {
 440       int v6only = 1;
 441       CHECK(0 == setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY,
 442                             &v6only, sizeof(v6only)));
 443     }
 444
 445     // Bind to the socket
 446     if (fsp::bind(s, res->ai_addr, socklen_t(res->ai_addrlen)) != 0) {
 447       folly::throwSystemError(
 448           errno,
 449           "failed to bind to async server socket for port ",
 450           SocketAddress::getPortFrom(res->ai_addr),
 451           " family ",
 452           SocketAddress::getFamilyNameFrom(res->ai_addr, "<unknown>"));
 453     }
 454
 455 #if __linux__
 456     if (noTransparentTls_) {
 457       // Ignore return value, errors are ok
 458       setsockopt(s, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
 459     }
 460 #endif
 461
 462     SocketAddress address;
 463     address.setFromLocalAddress(s);
 464
 465     sockets_.emplace_back(eventBase_, s, this, address.getFamily());
 466   };
 467
 468   const int kNumTries = 25;
 469   for (int tries = 1; true; tries++) {
 470     // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
 471     // should return IPv6 first and then IPv4 addresses, but glibc's
 472     // getaddrinfo(nullptr) with AI_PASSIVE returns:
 473     // - 0.0.0.0 (IPv4-only)
 474     // - :: (IPv6+IPv4) in this order
 475     // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
 476     for (struct addrinfo* res = res0; res; res = res->ai_next) {
 477       if (res->ai_family == AF_INET6) {
 478         setupAddress(res);
 479       }
 480     }
 481
 482     // If port == 0, then we should try to bind to the same port on ipv4 and
 483     // ipv6.  So if we did bind to ipv6, figure out that port and use it.
 484     if (sockets_.size() == 1 && port == 0) {
 485       SocketAddress address;
 486       address.setFromLocalAddress(sockets_.back().socket_);
 487       snprintf(sport, sizeof(sport), "%u", address.getPort());
 488       freeaddrinfo(res0);
 489       CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
 490     }
 491
 492     try {
 493       for (struct addrinfo* res = res0; res; res = res->ai_next) {
 494         if (res->ai_family != AF_INET6) {
 495           setupAddress(res);
 496         }
 497       }
 498     } catch (const std::system_error&) {
 499       // If we can't bind to the same port on ipv4 as ipv6 when using
 500       // port=0 then we will retry again before giving up after
 501       // kNumTries attempts.  We do this by closing the sockets that
 502       // were opened, then restarting from scratch.
 503       if (port == 0 && !sockets_.empty() && tries != kNumTries) {
 504         for (const auto& socket : sockets_) {
 505           if (socket.socket_ <= 0) {
 506             continue;
 507           } else if (shutdownSocketSet_) {
 508             shutdownSocketSet_->close(socket.socket_);
 509           } else {
 510             closeNoInt(socket.socket_);
 511           }
 512         }
 513         sockets_.clear();
 514         snprintf(sport, sizeof(sport), "%u", port);
 515         freeaddrinfo(res0);
 516         CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
 517         continue;
 518       }
 519
 520       throw;
 521     }
 522
 523     break;
 524   }
 525
 526   if (sockets_.size() == 0) {
 527     throw std::runtime_error(
 528         "did not bind any async server socket for port");
 529   }
 530 }
 531
 532 void AsyncServerSocket::listen(int backlog) {
 533   if (eventBase_) {
 534     eventBase_->dcheckIsInEventBaseThread();
 535   }
 536
 537   // Start listening
 538   for (auto& handler : sockets_) {
 539     if (fsp::listen(handler.socket_, backlog) == -1) {
 540       folly::throwSystemError(errno,
 541                                     "failed to listen on async server socket");
 542     }
 543   }
 544 }
 545
 546 void AsyncServerSocket::getAddress(SocketAddress* addressReturn) const {
 547   CHECK(sockets_.size() >= 1);
 548   VLOG_IF(2, sockets_.size() > 1)
 549     << "Warning: getAddress() called and multiple addresses available ("
 550     << sockets_.size() << "). Returning only the first one.";
 551
 552   addressReturn->setFromLocalAddress(sockets_[0].socket_);
 553 }
 554
 555 std::vector<SocketAddress> AsyncServerSocket::getAddresses()
 556     const {
 557   CHECK(sockets_.size() >= 1);
 558   auto tsaVec = std::vector<SocketAddress>(sockets_.size());
 559   auto tsaIter = tsaVec.begin();
 560   for (const auto& socket : sockets_) {
 561     (tsaIter++)->setFromLocalAddress(socket.socket_);
 562   };
 563   return tsaVec;
 564 }
 565
 566 void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
 567                                            EventBase *eventBase,
 568                                            uint32_t maxAtOnce) {
 569   if (eventBase_) {
 570     eventBase_->dcheckIsInEventBaseThread();
 571   }
 572
 573   // If this is the first accept callback and we are supposed to be accepting,
 574   // start accepting once the callback is installed.
 575   bool runStartAccepting = accepting_ && callbacks_.empty();
 576
 577   callbacks_.emplace_back(callback, eventBase);
 578
 579   SCOPE_SUCCESS {
 580     // If this is the first accept callback and we are supposed to be accepting,
 581     // start accepting.
 582     if (runStartAccepting) {
 583       startAccepting();
 584     }
 585   };
 586
 587   if (!eventBase) {
 588     // Run in AsyncServerSocket's eventbase; notify that we are
 589     // starting to accept connections
 590     callback->acceptStarted();
 591     return;
 592   }
 593
 594   // Start the remote acceptor.
 595   //
 596   // It would be nice if we could avoid starting the remote acceptor if
 597   // eventBase == eventBase_.  However, that would cause issues if
 598   // detachEventBase() and attachEventBase() were ever used to change the
 599   // primary EventBase for the server socket.  Therefore we require the caller
 600   // to specify a nullptr EventBase if they want to ensure that the callback is
 601   // always invoked in the primary EventBase, and to be able to invoke that
 602   // callback more efficiently without having to use a notification queue.
 603   RemoteAcceptor* acceptor = nullptr;
 604   try {
 605     acceptor = new RemoteAcceptor(callback, connectionEventCallback_);
 606     acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
 607   } catch (...) {
 608     callbacks_.pop_back();
 609     delete acceptor;
 610     throw;
 611   }
 612   callbacks_.back().consumer = acceptor;
 613 }
 614
 615 void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
 616                                               EventBase *eventBase) {
 617   if (eventBase_) {
 618     eventBase_->dcheckIsInEventBaseThread();
 619   }
 620
 621   // Find the matching AcceptCallback.
 622   // We just do a simple linear search; we don't expect removeAcceptCallback()
 623   // to be called frequently, and we expect there to only be a small number of
 624   // callbacks anyway.
 625   std::vector<CallbackInfo>::iterator it = callbacks_.begin();
 626   uint32_t n = 0;
 627   while (true) {
 628     if (it == callbacks_.end()) {
 629       throw std::runtime_error("AsyncServerSocket::removeAcceptCallback(): "
 630                               "accept callback not found");
 631     }
 632     if (it->callback == callback &&
 633         (it->eventBase == eventBase || eventBase == nullptr)) {
 634       break;
 635     }
 636     ++it;
 637     ++n;
 638   }
 639
 640   // Remove this callback from callbacks_.
 641   //
 642   // Do this before invoking the acceptStopped() callback, in case
 643   // acceptStopped() invokes one of our methods that examines callbacks_.
 644   //
 645   // Save a copy of the CallbackInfo first.
 646   CallbackInfo info(*it);
 647   callbacks_.erase(it);
 648   if (n < callbackIndex_) {
 649     // We removed an element before callbackIndex_.  Move callbackIndex_ back
 650     // one step, since things after n have been shifted back by 1.
 651     --callbackIndex_;
 652   } else {
 653     // We removed something at or after callbackIndex_.
 654     // If we removed the last element and callbackIndex_ was pointing at it,
 655     // we need to reset callbackIndex_ to 0.
 656     if (callbackIndex_ >= callbacks_.size()) {
 657       callbackIndex_ = 0;
 658     }
 659   }
 660
 661   if (info.consumer) {
 662     // consumer could be nullptr is we run callbacks in primary event
 663     // base
 664     DCHECK(info.eventBase);
 665     info.consumer->stop(info.eventBase, info.callback);
 666   } else {
 667     // callback invoked in the primary event base, just call directly
 668     DCHECK(info.callback);
 669     callback->acceptStopped();
 670   }
 671
 672   // If we are supposed to be accepting but the last accept callback
 673   // was removed, unregister for events until a callback is added.
 674   if (accepting_ && callbacks_.empty()) {
 675     for (auto& handler : sockets_) {
 676       handler.unregisterHandler();
 677     }
 678   }
 679 }
 680
 681 void AsyncServerSocket::startAccepting() {
 682   if (eventBase_) {
 683     eventBase_->dcheckIsInEventBaseThread();
 684   }
 685
 686   accepting_ = true;
 687   if (callbacks_.empty()) {
 688     // We can't actually begin accepting if no callbacks are defined.
 689     // Wait until a callback is added to start accepting.
 690     return;
 691   }
 692
 693   for (auto& handler : sockets_) {
 694     if (!handler.registerHandler(
 695           EventHandler::READ | EventHandler::PERSIST)) {
 696       throw std::runtime_error("failed to register for accept events");
 697     }
 698   }
 699 }
 700
 701 void AsyncServerSocket::pauseAccepting() {
 702   if (eventBase_) {
 703     eventBase_->dcheckIsInEventBaseThread();
 704   }
 705   accepting_ = false;
 706   for (auto& handler : sockets_) {
 707    handler. unregisterHandler();
 708   }
 709
 710   // If we were in the accept backoff state, disable the backoff timeout
 711   if (backoffTimeout_) {
 712     backoffTimeout_->cancelTimeout();
 713   }
 714 }
 715
 716 int AsyncServerSocket::createSocket(int family) {
 717   int fd = fsp::socket(family, SOCK_STREAM, 0);
 718   if (fd == -1) {
 719     folly::throwSystemError(errno, "error creating async server socket");
 720   }
 721
 722   try {
 723     setupSocket(fd, family);
 724   } catch (...) {
 725     closeNoInt(fd);
 726     throw;
 727   }
 728   return fd;
 729 }
 730
 731 void AsyncServerSocket::setupSocket(int fd, int family) {
 732   // Put the socket in non-blocking mode
 733   if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
 734     folly::throwSystemError(errno,
 735                             "failed to put socket in non-blocking mode");
 736   }
 737
 738   // Set reuseaddr to avoid 2MSL delay on server restart
 739   int one = 1;
 740   if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
 741     // This isn't a fatal error; just log an error message and continue
 742     LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
 743   }
 744
 745   // Set reuseport to support multiple accept threads
 746   int zero = 0;
 747   if (reusePortEnabled_ &&
 748       setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
 749     LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
 750                << strerror(errno);
 751 #ifdef WIN32
 752     folly::throwSystemError(errno, "failed to bind to the async server socket");
 753 #else
 754     SocketAddress address;
 755     address.setFromLocalAddress(fd);
 756     folly::throwSystemError(errno,
 757                             "failed to bind to async server socket: " +
 758                             address.describe());
 759 #endif
 760   }
 761
 762   // Set keepalive as desired
 763   if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
 764                  (keepAliveEnabled_) ? &one : &zero, sizeof(int)) != 0) {
 765     LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: " <<
 766             strerror(errno);
 767   }
 768
 769   // Setup FD_CLOEXEC flag
 770   if (closeOnExec_ &&
 771       (-1 == folly::setCloseOnExec(fd, closeOnExec_))) {
 772     LOG(ERROR) << "failed to set FD_CLOEXEC on async server socket: " <<
 773             strerror(errno);
 774   }
 775
 776   // Set TCP nodelay if available, MAC OS X Hack
 777   // See http://lists.danga.com/pipermail/memcached/2005-March/001240.html
 778 #ifndef TCP_NOPUSH
 779   if (family != AF_UNIX) {
 780     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0) {
 781       // This isn't a fatal error; just log an error message and continue
 782       LOG(ERROR) << "failed to set TCP_NODELAY on async server socket: " <<
 783               strerror(errno);
 784     }
 785   }
 786 #endif
 787
 788 #if FOLLY_ALLOW_TFO
 789   if (tfo_ && detail::tfo_enable(fd, tfoMaxQueueSize_) != 0) {
 790     // This isn't a fatal error; just log an error message and continue
 791     LOG(WARNING) << "failed to set TCP_FASTOPEN on async server socket: "
 792                  << folly::errnoStr(errno);
 793   }
 794 #endif
 795
 796   if (shutdownSocketSet_) {
 797     shutdownSocketSet_->add(fd);
 798   }
 799 }
 800
 801 void AsyncServerSocket::handlerReady(uint16_t /* events */,
 802                                      int fd,
 803                                      sa_family_t addressFamily) noexcept {
 804   assert(!callbacks_.empty());
 805   DestructorGuard dg(this);
 806
 807   // Only accept up to maxAcceptAtOnce_ connections at a time,
 808   // to avoid starving other I/O handlers using this EventBase.
 809   for (uint32_t n = 0; n < maxAcceptAtOnce_; ++n) {
 810     SocketAddress address;
 811
 812     sockaddr_storage addrStorage;
 813     socklen_t addrLen = sizeof(addrStorage);
 814     sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 815
 816     // In some cases, accept() doesn't seem to update these correctly.
 817     saddr->sa_family = addressFamily;
 818     if (addressFamily == AF_UNIX) {
 819       addrLen = sizeof(struct sockaddr_un);
 820     }
 821
 822     // Accept a new client socket
 823 #ifdef SOCK_NONBLOCK
 824     int clientSocket = accept4(fd, saddr, &addrLen, SOCK_NONBLOCK);
 825 #else
 826     int clientSocket = accept(fd, saddr, &addrLen);
 827 #endif
 828
 829     address.setFromSockaddr(saddr, addrLen);
 830
 831     if (clientSocket >= 0 && connectionEventCallback_) {
 832       connectionEventCallback_->onConnectionAccepted(clientSocket, address);
 833     }
 834
 835     std::chrono::time_point<std::chrono::steady_clock> nowMs =
 836       std::chrono::steady_clock::now();
 837     auto timeSinceLastAccept = std::max<int64_t>(
 838       0,
 839       nowMs.time_since_epoch().count() -
 840       lastAccepTimestamp_.time_since_epoch().count());
 841     lastAccepTimestamp_ = nowMs;
 842     if (acceptRate_ < 1) {
 843       acceptRate_ *= 1 + acceptRateAdjustSpeed_ * timeSinceLastAccept;
 844       if (acceptRate_ >= 1) {
 845         acceptRate_ = 1;
 846       } else if (rand() > acceptRate_ * RAND_MAX) {
 847         ++numDroppedConnections_;
 848         if (clientSocket >= 0) {
 849           closeNoInt(clientSocket);
 850           if (connectionEventCallback_) {
 851             connectionEventCallback_->onConnectionDropped(clientSocket,
 852                                                           address);
 853           }
 854         }
 855         continue;
 856       }
 857     }
 858
 859     if (clientSocket < 0) {
 860       if (errno == EAGAIN) {
 861         // No more sockets to accept right now.
 862         // Check for this code first, since it's the most common.
 863         return;
 864       } else if (errno == EMFILE || errno == ENFILE) {
 865         // We're out of file descriptors.  Perhaps we're accepting connections
 866         // too quickly. Pause accepting briefly to back off and give the server
 867         // a chance to recover.
 868         LOG(ERROR) << "accept failed: out of file descriptors; entering accept "
 869                 "back-off state";
 870         enterBackoff();
 871
 872         // Dispatch the error message
 873         dispatchError("accept() failed", errno);
 874       } else {
 875         dispatchError("accept() failed", errno);
 876       }
 877       if (connectionEventCallback_) {
 878         connectionEventCallback_->onConnectionAcceptError(errno);
 879       }
 880       return;
 881     }
 882
 883 #ifndef SOCK_NONBLOCK
 884     // Explicitly set the new connection to non-blocking mode
 885     if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
 886       closeNoInt(clientSocket);
 887       dispatchError("failed to set accepted socket to non-blocking mode",
 888                     errno);
 889       if (connectionEventCallback_) {
 890         connectionEventCallback_->onConnectionDropped(clientSocket, address);
 891       }
 892       return;
 893     }
 894 #endif
 895
 896     // Inform the callback about the new connection
 897     dispatchSocket(clientSocket, std::move(address));
 898
 899     // If we aren't accepting any more, break out of the loop
 900     if (!accepting_ || callbacks_.empty()) {
 901       break;
 902     }
 903   }
 904 }
 905
 906 void AsyncServerSocket::dispatchSocket(int socket,
 907                                         SocketAddress&& address) {
 908   uint32_t startingIndex = callbackIndex_;
 909
 910   // Short circuit if the callback is in the primary EventBase thread
 911
 912   CallbackInfo *info = nextCallback();
 913   if (info->eventBase == nullptr) {
 914     info->callback->connectionAccepted(socket, address);
 915     return;
 916   }
 917
 918   const SocketAddress addr(address);
 919   // Create a message to send over the notification queue
 920   QueueMessage msg;
 921   msg.type = MessageType::MSG_NEW_CONN;
 922   msg.address = std::move(address);
 923   msg.fd = socket;
 924
 925   // Loop until we find a free queue to write to
 926   while (true) {
 927     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
 928       if (connectionEventCallback_) {
 929         connectionEventCallback_->onConnectionEnqueuedForAcceptorCallback(
 930             socket,
 931             addr);
 932       }
 933       // Success! return.
 934       return;
 935     }
 936
 937     // We couldn't add to queue.  Fall through to below
 938
 939     ++numDroppedConnections_;
 940     if (acceptRateAdjustSpeed_ > 0) {
 941       // aggressively decrease accept rate when in trouble
 942       static const double kAcceptRateDecreaseSpeed = 0.1;
 943       acceptRate_ *= 1 - kAcceptRateDecreaseSpeed;
 944     }
 945
 946
 947     if (callbackIndex_ == startingIndex) {
 948       // The notification queue was full
 949       // We can't really do anything at this point other than close the socket.
 950       //
 951       // This should only happen if a user's service is behaving extremely
 952       // badly and none of the EventBase threads are looping fast enough to
 953       // process the incoming connections.  If the service is overloaded, it
 954       // should use pauseAccepting() to temporarily back off accepting new
 955       // connections, before they reach the point where their threads can't
 956       // even accept new messages.
 957       LOG(ERROR) << "failed to dispatch newly accepted socket:"
 958                  << " all accept callback queues are full";
 959       closeNoInt(socket);
 960       if (connectionEventCallback_) {
 961         connectionEventCallback_->onConnectionDropped(socket, addr);
 962       }
 963       return;
 964     }
 965
 966     info = nextCallback();
 967   }
 968 }
 969
 970 void AsyncServerSocket::dispatchError(const char *msgstr, int errnoValue) {
 971   uint32_t startingIndex = callbackIndex_;
 972   CallbackInfo *info = nextCallback();
 973
 974   // Create a message to send over the notification queue
 975   QueueMessage msg;
 976   msg.type = MessageType::MSG_ERROR;
 977   msg.err = errnoValue;
 978   msg.msg = std::move(msgstr);
 979
 980   while (true) {
 981     // Short circuit if the callback is in the primary EventBase thread
 982     if (info->eventBase == nullptr) {
 983       std::runtime_error ex(
 984         std::string(msgstr) +  folly::to<std::string>(errnoValue));
 985       info->callback->acceptError(ex);
 986       return;
 987     }
 988
 989     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
 990       return;
 991     }
 992     // Fall through and try another callback
 993
 994     if (callbackIndex_ == startingIndex) {
 995       // The notification queues for all of the callbacks were full.
 996       // We can't really do anything at this point.
 997       LOG(ERROR) << "failed to dispatch accept error: all accept callback "
 998         "queues are full: error msg:  " <<
 999         msg.msg.c_str() << errnoValue;
1000       return;
1001     }
1002     info = nextCallback();
1003   }
1004 }
1005
1006 void AsyncServerSocket::enterBackoff() {
1007   // If this is the first time we have entered the backoff state,
1008   // allocate backoffTimeout_.
1009   if (backoffTimeout_ == nullptr) {
1010     try {
1011       backoffTimeout_ = new BackoffTimeout(this);
1012     } catch (const std::bad_alloc&) {
1013       // Man, we couldn't even allocate the timer to re-enable accepts.
1014       // We must be in pretty bad shape.  Don't pause accepting for now,
1015       // since we won't be able to re-enable ourselves later.
1016       LOG(ERROR) << "failed to allocate AsyncServerSocket backoff"
1017                  << " timer; unable to temporarly pause accepting";
1018       if (connectionEventCallback_) {
1019         connectionEventCallback_->onBackoffError();
1020       }
1021       return;
1022     }
1023   }
1024
1025   // For now, we simply pause accepting for 1 second.
1026   //
1027   // We could add some smarter backoff calculation here in the future.  (e.g.,
1028   // start sleeping for longer if we keep hitting the backoff frequently.)
1029   // Typically the user needs to figure out why the server is overloaded and
1030   // fix it in some other way, though.  The backoff timer is just a simple
1031   // mechanism to try and give the connection processing code a little bit of
1032   // breathing room to catch up, and to avoid just spinning and failing to
1033   // accept over and over again.
1034   const uint32_t timeoutMS = 1000;
1035   if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
1036     LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
1037                << "unable to temporarly pause accepting";
1038     if (connectionEventCallback_) {
1039       connectionEventCallback_->onBackoffError();
1040     }
1041     return;
1042   }
1043
1044   // The backoff timer is scheduled to re-enable accepts.
1045   // Go ahead and disable accepts for now.  We leave accepting_ set to true,
1046   // since that tracks the desired state requested by the user.
1047   for (auto& handler : sockets_) {
1048     handler.unregisterHandler();
1049   }
1050   if (connectionEventCallback_) {
1051     connectionEventCallback_->onBackoffStarted();
1052   }
1053 }
1054
1055 void AsyncServerSocket::backoffTimeoutExpired() {
1056   // accepting_ should still be true.
1057   // If pauseAccepting() was called while in the backoff state it will cancel
1058   // the backoff timeout.
1059   assert(accepting_);
1060   // We can't be detached from the EventBase without being paused
1061   assert(eventBase_ != nullptr);
1062   eventBase_->dcheckIsInEventBaseThread();
1063
1064   // If all of the callbacks were removed, we shouldn't re-enable accepts
1065   if (callbacks_.empty()) {
1066     if (connectionEventCallback_) {
1067       connectionEventCallback_->onBackoffEnded();
1068     }
1069     return;
1070   }
1071
1072   // Register the handler.
1073   for (auto& handler : sockets_) {
1074     if (!handler.registerHandler(
1075           EventHandler::READ | EventHandler::PERSIST)) {
1076       // We're hosed.  We could just re-schedule backoffTimeout_ to
1077       // re-try again after a little bit.  However, we don't want to
1078       // loop retrying forever if we can't re-enable accepts.  Just
1079       // abort the entire program in this state; things are really bad
1080       // and restarting the entire server is probably the best remedy.
1081       LOG(ERROR)
1082         << "failed to re-enable AsyncServerSocket accepts after backoff; "
1083         << "crashing now";
1084       abort();
1085     }
1086   }
1087   if (connectionEventCallback_) {
1088     connectionEventCallback_->onBackoffEnded();
1089   }
1090 }
1091
1092 } // namespace folly