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