folly/io/async/AsyncServerSocket.cpp

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