X-Git-Url: http://plrg.eecs.uci.edu/git/?a=blobdiff_plain;f=folly%2Fio%2Fasync%2FAsyncServerSocket.cpp;h=0d3684ed0e3ae5ae820932a2323d52bfb0d3faa2;hb=54b16a23784d48f1a3b56ff0c33bdf7fdfe46355;hp=e05f37076864c8612df6527a14cd73ab095eb4d6;hpb=ad4a9ff72a9bc385a01f075f977f29eb805fd127;p=folly.git diff --git a/folly/io/async/AsyncServerSocket.cpp b/folly/io/async/AsyncServerSocket.cpp index e05f3707..0d3684ed 100644 --- a/folly/io/async/AsyncServerSocket.cpp +++ b/folly/io/async/AsyncServerSocket.cpp @@ -1,5 +1,5 @@ /* - * Copyright 2014 Facebook, Inc. + * Copyright 2017 Facebook, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,20 +21,31 @@ #include #include +#include #include +#include +#include #include #include +#include +#include +#include #include -#include -#include #include -#include #include -#include + +namespace fsp = folly::portability::sockets; namespace folly { +static constexpr bool msgErrQueueSupported = +#ifdef FOLLY_HAVE_MSG_ERRQUEUE + true; +#else + false; +#endif // FOLLY_HAVE_MSG_ERRQUEUE + const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce; const uint32_t AsyncServerSocket::kDefaultCallbackAcceptAtOnce; const uint32_t AsyncServerSocket::kDefaultMaxMessagesInQueue; @@ -44,15 +55,17 @@ int setCloseOnExec(int fd, int value) { int old_flags = fcntl(fd, F_GETFD, 0); // If reading the flags failed, return error indication now - if (old_flags < 0) + if (old_flags < 0) { return -1; + } // Set just the flag we want to set int new_flags; - if (value != 0) + if (value != 0) { new_flags = old_flags | FD_CLOEXEC; - else + } else { new_flags = old_flags & ~FD_CLOEXEC; + } // Store modified flag word in the descriptor return fcntl(fd, F_SETFD, new_flags); @@ -86,11 +99,14 @@ void AsyncServerSocket::RemoteAcceptor::stop( } void AsyncServerSocket::RemoteAcceptor::messageAvailable( - QueueMessage&& msg) { - + QueueMessage&& msg) noexcept { switch (msg.type) { case MessageType::MSG_NEW_CONN: { + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionDequeuedByAcceptorCallback( + msg.fd, msg.address); + } callback_->connectionAccepted(msg.fd, msg.address); break; } @@ -116,13 +132,12 @@ void AsyncServerSocket::RemoteAcceptor::messageAvailable( */ class AsyncServerSocket::BackoffTimeout : public AsyncTimeout { public: - BackoffTimeout(AsyncServerSocket* socket) - : AsyncTimeout(socket->getEventBase()), - socket_(socket) {} + // Disallow copy, move, and default constructors. + BackoffTimeout(BackoffTimeout&&) = delete; + explicit BackoffTimeout(AsyncServerSocket* socket) + : AsyncTimeout(socket->getEventBase()), socket_(socket) {} - virtual void timeoutExpired() noexcept { - socket_->backoffTimeoutExpired(); - } + void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); } private: AsyncServerSocket* socket_; @@ -133,37 +148,42 @@ class AsyncServerSocket::BackoffTimeout : public AsyncTimeout { */ AsyncServerSocket::AsyncServerSocket(EventBase* eventBase) -: eventBase_(eventBase), - accepting_(false), - maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce), - maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue), - acceptRateAdjustSpeed_(0), - acceptRate_(1), - lastAccepTimestamp_(std::chrono::steady_clock::now()), - numDroppedConnections_(0), - callbackIndex_(0), - backoffTimeout_(nullptr), - callbacks_(), - keepAliveEnabled_(true), - closeOnExec_(true), - shutdownSocketSet_(nullptr) { -} - -void AsyncServerSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) { - if (shutdownSocketSet_ == newSS) { + : eventBase_(eventBase), + accepting_(false), + maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce), + maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue), + acceptRateAdjustSpeed_(0), + acceptRate_(1), + lastAccepTimestamp_(std::chrono::steady_clock::now()), + numDroppedConnections_(0), + callbackIndex_(0), + backoffTimeout_(nullptr), + callbacks_(), + keepAliveEnabled_(true), + closeOnExec_(true) {} + +void AsyncServerSocket::setShutdownSocketSet( + const std::weak_ptr& wNewSS) { + const auto newSS = wNewSS.lock(); + const auto shutdownSocketSet = wShutdownSocketSet_.lock(); + + if (shutdownSocketSet == newSS) { return; } - if (shutdownSocketSet_) { + + if (shutdownSocketSet) { for (auto& h : sockets_) { - shutdownSocketSet_->remove(h.socket_); + shutdownSocketSet->remove(h.socket_); } } - shutdownSocketSet_ = newSS; - if (shutdownSocketSet_) { + + if (newSS) { for (auto& h : sockets_) { - shutdownSocketSet_->add(h.socket_); + newSS->add(h.socket_); } } + + wShutdownSocketSet_ = wNewSS; } AsyncServerSocket::~AsyncServerSocket() { @@ -176,15 +196,22 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { VLOG(10) << "AsyncServerSocket::stopAccepting " << this << handler.socket_; } - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } - // When destroy is called, unregister and close the socket immediately + // When destroy is called, unregister and close the socket immediately. accepting_ = false; - for (auto& handler : sockets_) { + // Close the sockets in reverse order as they were opened to avoid + // the condition where another process concurrently tries to open + // the same port, succeed to bind the first socket but fails on the + // second because it hasn't been closed yet. + for (; !sockets_.empty(); sockets_.pop_back()) { + auto& handler = sockets_.back(); handler.unregisterHandler(); - if (shutdownSocketSet_) { - shutdownSocketSet_->close(handler.socket_); + if (const auto shutdownSocketSet = wShutdownSocketSet_.lock()) { + shutdownSocketSet->close(handler.socket_); } else if (shutdownFlags >= 0) { result = shutdownNoInt(handler.socket_, shutdownFlags); pendingCloseSockets_.push_back(handler.socket_); @@ -192,7 +219,6 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { closeNoInt(handler.socket_); } } - sockets_.clear(); // Destroy the backoff timout. This will cancel it if it is running. delete backoffTimeout_; @@ -209,7 +235,14 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { for (std::vector::iterator it = callbacksCopy.begin(); it != callbacksCopy.end(); ++it) { - it->consumer->stop(it->eventBase, it->callback); + // consumer may not be set if we are running in primary event base + if (it->consumer) { + DCHECK(it->eventBase); + it->consumer->stop(it->eventBase, it->callback); + } else { + DCHECK(it->callback); + it->callback->acceptStopped(); + } } return result; @@ -227,7 +260,7 @@ void AsyncServerSocket::destroy() { void AsyncServerSocket::attachEventBase(EventBase *eventBase) { assert(eventBase_ == nullptr); - assert(eventBase->isInEventBaseThread()); + eventBase->dcheckIsInEventBaseThread(); eventBase_ = eventBase; for (auto& handler : sockets_) { @@ -237,7 +270,7 @@ void AsyncServerSocket::attachEventBase(EventBase *eventBase) { void AsyncServerSocket::detachEventBase() { assert(eventBase_ != nullptr); - assert(eventBase_->isInEventBaseThread()); + eventBase_->dcheckIsInEventBaseThread(); assert(!accepting_); eventBase_ = nullptr; @@ -247,7 +280,9 @@ void AsyncServerSocket::detachEventBase() { } void AsyncServerSocket::useExistingSockets(const std::vector& fds) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } if (sockets_.size() > 0) { throw std::invalid_argument( @@ -263,9 +298,15 @@ void AsyncServerSocket::useExistingSockets(const std::vector& fds) { SocketAddress address; address.setFromLocalAddress(fd); - setupSocket(fd); - sockets_.push_back( - ServerEventHandler(eventBase_, fd, this, address.getFamily())); +#if __linux__ + if (noTransparentTls_) { + // Ignore return value, errors are ok + setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0); + } +#endif + + setupSocket(fd, address.getFamily()); + sockets_.emplace_back(eventBase_, fd, this, address.getFamily()); sockets_.back().changeHandlerFD(fd); } } @@ -281,7 +322,8 @@ void AsyncServerSocket::bindSocket( sockaddr_storage addrStorage; address.getAddress(&addrStorage); sockaddr* saddr = reinterpret_cast(&addrStorage); - if (::bind(fd, saddr, address.getActualSize()) != 0) { + + if (fsp::bind(fd, saddr, address.getActualSize()) != 0) { if (!isExistingSocket) { closeNoInt(fd); } @@ -290,15 +332,35 @@ void AsyncServerSocket::bindSocket( address.describe()); } +#if __linux__ + if (noTransparentTls_) { + // Ignore return value, errors are ok + setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0); + } +#endif + // If we just created this socket, update the EventHandler and set socket_ if (!isExistingSocket) { - sockets_.push_back( - ServerEventHandler(eventBase_, fd, this, address.getFamily())); + sockets_.emplace_back(eventBase_, fd, this, address.getFamily()); } } +bool AsyncServerSocket::setZeroCopy(bool enable) { + if (msgErrQueueSupported) { + int fd = getSocket(); + int val = enable ? 1 : 0; + int ret = setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, &val, sizeof(val)); + + return (0 == ret); + } + + return false; +} + void AsyncServerSocket::bind(const SocketAddress& address) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // useExistingSocket() may have been called to initialize socket_ already. // However, in the normal case we need to create a new socket now. @@ -346,16 +408,19 @@ void AsyncServerSocket::bind( } void AsyncServerSocket::bind(uint16_t port) { - struct addrinfo hints, *res, *res0; + struct addrinfo hints, *res0; char sport[sizeof("65536")]; memset(&hints, 0, sizeof(hints)); hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM; - hints.ai_flags = AI_PASSIVE; + hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV; snprintf(sport, sizeof(sport), "%u", port); - if (getaddrinfo(nullptr, sport, &hints, &res0)) { + // On Windows the value we need to pass to bind to all available + // addresses is an empty string. Everywhere else, it's nullptr. + constexpr const char* kWildcardNode = kIsWindows ? "" : nullptr; + if (getaddrinfo(kWildcardNode, sport, &hints, &res0)) { throw std::invalid_argument( "Attempted to bind address to socket with " "bad getaddrinfo"); @@ -364,7 +429,7 @@ void AsyncServerSocket::bind(uint16_t port) { SCOPE_EXIT { freeaddrinfo(res0); }; auto setupAddress = [&] (struct addrinfo* res) { - int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + int s = fsp::socket(res->ai_family, res->ai_socktype, res->ai_protocol); // IPv6/IPv4 may not be supported by the kernel if (s < 0 && errno == EAFNOSUPPORT) { return; @@ -372,7 +437,7 @@ void AsyncServerSocket::bind(uint16_t port) { CHECK_GE(s, 0); try { - setupSocket(s); + setupSocket(s, res->ai_family); } catch (...) { closeNoInt(s); throw; @@ -384,21 +449,30 @@ void AsyncServerSocket::bind(uint16_t port) { &v6only, sizeof(v6only))); } - SocketAddress address; - address.setFromLocalAddress(s); - - sockets_.push_back( - ServerEventHandler(eventBase_, s, this, address.getFamily())); - // Bind to the socket - if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) { + if (fsp::bind(s, res->ai_addr, socklen_t(res->ai_addrlen)) != 0) { folly::throwSystemError( - errno, - "failed to bind to async server socket for port"); + errno, + "failed to bind to async server socket for port ", + SocketAddress::getPortFrom(res->ai_addr), + " family ", + SocketAddress::getFamilyNameFrom(res->ai_addr, "")); + } + +#if __linux__ + if (noTransparentTls_) { + // Ignore return value, errors are ok + setsockopt(s, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0); } +#endif + + SocketAddress address; + address.setFromLocalAddress(s); + + sockets_.emplace_back(eventBase_, s, this, address.getFamily()); }; - const int kNumTries = 5; + const int kNumTries = 25; for (int tries = 1; true; tries++) { // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo // should return IPv6 first and then IPv4 addresses, but glibc's @@ -406,15 +480,14 @@ void AsyncServerSocket::bind(uint16_t port) { // - 0.0.0.0 (IPv4-only) // - :: (IPv6+IPv4) in this order // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981 - for (res = res0; res; res = res->ai_next) { + for (struct addrinfo* res = res0; res; res = res->ai_next) { if (res->ai_family == AF_INET6) { setupAddress(res); } } // If port == 0, then we should try to bind to the same port on ipv4 and - // ipv6. So if we did bind to ipv6, figure out that port and use it, - // except for the last attempt when we just use any port available. + // ipv6. So if we did bind to ipv6, figure out that port and use it. if (sockets_.size() == 1 && port == 0) { SocketAddress address; address.setFromLocalAddress(sockets_.back().socket_); @@ -424,21 +497,23 @@ void AsyncServerSocket::bind(uint16_t port) { } try { - for (res = res0; res; res = res->ai_next) { + for (struct addrinfo* res = res0; res; res = res->ai_next) { if (res->ai_family != AF_INET6) { setupAddress(res); } } - } catch (const std::system_error& e) { - // if we can't bind to the same port on ipv4 as ipv6 when using port=0 - // then we will try again another 2 times before giving up. We do this - // by closing the sockets that were opened, then redoing the whole thing + } catch (const std::system_error&) { + // If we can't bind to the same port on ipv4 as ipv6 when using + // port=0 then we will retry again before giving up after + // kNumTries attempts. We do this by closing the sockets that + // were opened, then restarting from scratch. if (port == 0 && !sockets_.empty() && tries != kNumTries) { for (const auto& socket : sockets_) { if (socket.socket_ <= 0) { continue; - } else if (shutdownSocketSet_) { - shutdownSocketSet_->close(socket.socket_); + } else if ( + const auto shutdownSocketSet = wShutdownSocketSet_.lock()) { + shutdownSocketSet->close(socket.socket_); } else { closeNoInt(socket.socket_); } @@ -449,6 +524,7 @@ void AsyncServerSocket::bind(uint16_t port) { CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0)); continue; } + throw; } @@ -462,11 +538,13 @@ void AsyncServerSocket::bind(uint16_t port) { } void AsyncServerSocket::listen(int backlog) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // Start listening for (auto& handler : sockets_) { - if (::listen(handler.socket_, backlog) == -1) { + if (fsp::listen(handler.socket_, backlog) == -1) { folly::throwSystemError(errno, "failed to listen on async server socket"); } @@ -496,18 +574,31 @@ std::vector AsyncServerSocket::getAddresses() void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback, EventBase *eventBase, uint32_t maxAtOnce) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // If this is the first accept callback and we are supposed to be accepting, // start accepting once the callback is installed. bool runStartAccepting = accepting_ && callbacks_.empty(); + callbacks_.emplace_back(callback, eventBase); + + SCOPE_SUCCESS { + // If this is the first accept callback and we are supposed to be accepting, + // start accepting. + if (runStartAccepting) { + startAccepting(); + } + }; + if (!eventBase) { - eventBase = eventBase_; // Run in AsyncServerSocket's eventbase + // Run in AsyncServerSocket's eventbase; notify that we are + // starting to accept connections + callback->acceptStarted(); + return; } - callbacks_.push_back(CallbackInfo(callback, eventBase)); - // Start the remote acceptor. // // It would be nice if we could avoid starting the remote acceptor if @@ -519,7 +610,7 @@ void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback, // callback more efficiently without having to use a notification queue. RemoteAcceptor* acceptor = nullptr; try { - acceptor = new RemoteAcceptor(callback); + acceptor = new RemoteAcceptor(callback, connectionEventCallback_); acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_); } catch (...) { callbacks_.pop_back(); @@ -527,17 +618,13 @@ void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback, throw; } callbacks_.back().consumer = acceptor; - - // If this is the first accept callback and we are supposed to be accepting, - // start accepting. - if (runStartAccepting) { - startAccepting(); - } } void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, EventBase *eventBase) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // Find the matching AcceptCallback. // We just do a simple linear search; we don't expect removeAcceptCallback() @@ -579,7 +666,16 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, } } - info.consumer->stop(info.eventBase, info.callback); + if (info.consumer) { + // consumer could be nullptr is we run callbacks in primary event + // base + DCHECK(info.eventBase); + info.consumer->stop(info.eventBase, info.callback); + } else { + // callback invoked in the primary event base, just call directly + DCHECK(info.callback); + callback->acceptStopped(); + } // If we are supposed to be accepting but the last accept callback // was removed, unregister for events until a callback is added. @@ -591,7 +687,9 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, } void AsyncServerSocket::startAccepting() { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } accepting_ = true; if (callbacks_.empty()) { @@ -609,7 +707,9 @@ void AsyncServerSocket::startAccepting() { } void AsyncServerSocket::pauseAccepting() { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } accepting_ = false; for (auto& handler : sockets_) { handler. unregisterHandler(); @@ -622,13 +722,13 @@ void AsyncServerSocket::pauseAccepting() { } int AsyncServerSocket::createSocket(int family) { - int fd = socket(family, SOCK_STREAM, 0); + int fd = fsp::socket(family, SOCK_STREAM, 0); if (fd == -1) { folly::throwSystemError(errno, "error creating async server socket"); } try { - setupSocket(fd); + setupSocket(fd, family); } catch (...) { closeNoInt(fd); throw; @@ -636,12 +736,7 @@ int AsyncServerSocket::createSocket(int family) { return fd; } -void AsyncServerSocket::setupSocket(int fd) { - // Get the address family - SocketAddress address; - address.setFromLocalAddress(fd); - auto family = address.getFamily(); - +void AsyncServerSocket::setupSocket(int fd, int family) { // Put the socket in non-blocking mode if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) { folly::throwSystemError(errno, @@ -661,9 +756,15 @@ void AsyncServerSocket::setupSocket(int fd) { setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) { LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket " << strerror(errno); +#ifdef WIN32 + folly::throwSystemError(errno, "failed to bind to the async server socket"); +#else + SocketAddress address; + address.setFromLocalAddress(fd); folly::throwSystemError(errno, "failed to bind to async server socket: " + address.describe()); +#endif } // Set keepalive as desired @@ -692,13 +793,22 @@ void AsyncServerSocket::setupSocket(int fd) { } #endif - if (shutdownSocketSet_) { - shutdownSocketSet_->add(fd); +#if FOLLY_ALLOW_TFO + if (tfo_ && detail::tfo_enable(fd, tfoMaxQueueSize_) != 0) { + // This isn't a fatal error; just log an error message and continue + LOG(WARNING) << "failed to set TCP_FASTOPEN on async server socket: " + << folly::errnoStr(errno); + } +#endif + + if (const auto shutdownSocketSet = wShutdownSocketSet_.lock()) { + shutdownSocketSet->add(fd); } } -void AsyncServerSocket::handlerReady( - uint16_t events, int fd, sa_family_t addressFamily) noexcept { +void AsyncServerSocket::handlerReady(uint16_t /* events */, + int fd, + sa_family_t addressFamily) noexcept { assert(!callbacks_.empty()); DestructorGuard dg(this); @@ -726,10 +836,14 @@ void AsyncServerSocket::handlerReady( address.setFromSockaddr(saddr, addrLen); + if (clientSocket >= 0 && connectionEventCallback_) { + connectionEventCallback_->onConnectionAccepted(clientSocket, address); + } + std::chrono::time_point nowMs = std::chrono::steady_clock::now(); - int64_t timeSinceLastAccept = std::max( - int64_t(0), + auto timeSinceLastAccept = std::max( + 0, nowMs.time_since_epoch().count() - lastAccepTimestamp_.time_since_epoch().count()); lastAccepTimestamp_ = nowMs; @@ -741,6 +855,10 @@ void AsyncServerSocket::handlerReady( ++numDroppedConnections_; if (clientSocket >= 0) { closeNoInt(clientSocket); + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionDropped(clientSocket, + address); + } } continue; } @@ -764,6 +882,9 @@ void AsyncServerSocket::handlerReady( } else { dispatchError("accept() failed", errno); } + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionAcceptError(errno); + } return; } @@ -773,6 +894,9 @@ void AsyncServerSocket::handlerReady( closeNoInt(clientSocket); dispatchError("failed to set accepted socket to non-blocking mode", errno); + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionDropped(clientSocket, address); + } return; } #endif @@ -799,6 +923,7 @@ void AsyncServerSocket::dispatchSocket(int socket, return; } + const SocketAddress addr(address); // Create a message to send over the notification queue QueueMessage msg; msg.type = MessageType::MSG_NEW_CONN; @@ -808,9 +933,14 @@ void AsyncServerSocket::dispatchSocket(int socket, // Loop until we find a free queue to write to while (true) { if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) { + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionEnqueuedForAcceptorCallback( + socket, + addr); + } // Success! return. return; - } + } // We couldn't add to queue. Fall through to below @@ -832,9 +962,12 @@ void AsyncServerSocket::dispatchSocket(int socket, // should use pauseAccepting() to temporarily back off accepting new // connections, before they reach the point where their threads can't // even accept new messages. - LOG(ERROR) << "failed to dispatch newly accepted socket:" - << " all accept callback queues are full"; + LOG_EVERY_N(ERROR, 100) << "failed to dispatch newly accepted socket:" + << " all accept callback queues are full"; closeNoInt(socket); + if (connectionEventCallback_) { + connectionEventCallback_->onConnectionDropped(socket, addr); + } return; } @@ -869,9 +1002,10 @@ void AsyncServerSocket::dispatchError(const char *msgstr, int errnoValue) { if (callbackIndex_ == startingIndex) { // The notification queues for all of the callbacks were full. // We can't really do anything at this point. - LOG(ERROR) << "failed to dispatch accept error: all accept callback " - "queues are full: error msg: " << - msg.msg.c_str() << errnoValue; + LOG_EVERY_N(ERROR, 100) + << "failed to dispatch accept error: all accept" + << " callback queues are full: error msg: " << msg.msg << ": " + << errnoValue; return; } info = nextCallback(); @@ -884,12 +1018,15 @@ void AsyncServerSocket::enterBackoff() { if (backoffTimeout_ == nullptr) { try { backoffTimeout_ = new BackoffTimeout(this); - } catch (const std::bad_alloc& ex) { + } catch (const std::bad_alloc&) { // Man, we couldn't even allocate the timer to re-enable accepts. // We must be in pretty bad shape. Don't pause accepting for now, // since we won't be able to re-enable ourselves later. LOG(ERROR) << "failed to allocate AsyncServerSocket backoff" << " timer; unable to temporarly pause accepting"; + if (connectionEventCallback_) { + connectionEventCallback_->onBackoffError(); + } return; } } @@ -907,6 +1044,9 @@ void AsyncServerSocket::enterBackoff() { if (!backoffTimeout_->scheduleTimeout(timeoutMS)) { LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;" << "unable to temporarly pause accepting"; + if (connectionEventCallback_) { + connectionEventCallback_->onBackoffError(); + } return; } @@ -916,6 +1056,9 @@ void AsyncServerSocket::enterBackoff() { for (auto& handler : sockets_) { handler.unregisterHandler(); } + if (connectionEventCallback_) { + connectionEventCallback_->onBackoffStarted(); + } } void AsyncServerSocket::backoffTimeoutExpired() { @@ -924,10 +1067,14 @@ void AsyncServerSocket::backoffTimeoutExpired() { // the backoff timeout. assert(accepting_); // We can't be detached from the EventBase without being paused - assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread()); + assert(eventBase_ != nullptr); + eventBase_->dcheckIsInEventBaseThread(); // If all of the callbacks were removed, we shouldn't re-enable accepts if (callbacks_.empty()) { + if (connectionEventCallback_) { + connectionEventCallback_->onBackoffEnded(); + } return; } @@ -946,8 +1093,9 @@ void AsyncServerSocket::backoffTimeoutExpired() { abort(); } } + if (connectionEventCallback_) { + connectionEventCallback_->onBackoffEnded(); + } } - - -} // folly +} // namespace folly