2 * Copyright 2015 Facebook, Inc.
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
8 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 #ifndef __STDC_FORMAT_MACROS
18 #define __STDC_FORMAT_MACROS
21 #include <folly/io/async/AsyncServerSocket.h>
23 #include <folly/FileUtil.h>
24 #include <folly/SocketAddress.h>
25 #include <folly/io/async/EventBase.h>
26 #include <folly/io/async/NotificationQueue.h>
30 #include <netinet/tcp.h>
32 #include <sys/socket.h>
33 #include <sys/types.h>
38 const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce;
39 const uint32_t AsyncServerSocket::kDefaultCallbackAcceptAtOnce;
40 const uint32_t AsyncServerSocket::kDefaultMaxMessagesInQueue;
42 int setCloseOnExec(int fd, int value) {
43 // Read the current flags
44 int old_flags = fcntl(fd, F_GETFD, 0);
46 // If reading the flags failed, return error indication now
50 // Set just the flag we want to set
53 new_flags = old_flags | FD_CLOEXEC;
55 new_flags = old_flags & ~FD_CLOEXEC;
57 // Store modified flag word in the descriptor
58 return fcntl(fd, F_SETFD, new_flags);
61 void AsyncServerSocket::RemoteAcceptor::start(
62 EventBase* eventBase, uint32_t maxAtOnce, uint32_t maxInQueue) {
63 setMaxReadAtOnce(maxAtOnce);
64 queue_.setMaxQueueSize(maxInQueue);
66 if (!eventBase->runInEventBaseThread([=](){
67 callback_->acceptStarted();
68 this->startConsuming(eventBase, &queue_);
70 throw std::invalid_argument("unable to start waiting on accept "
71 "notification queue in the specified "
76 void AsyncServerSocket::RemoteAcceptor::stop(
77 EventBase* eventBase, AcceptCallback* callback) {
78 if (!eventBase->runInEventBaseThread([=](){
79 callback->acceptStopped();
82 throw std::invalid_argument("unable to start waiting on accept "
83 "notification queue in the specified "
88 void AsyncServerSocket::RemoteAcceptor::messageAvailable(
92 case MessageType::MSG_NEW_CONN:
94 if (connectionEventCallback_) {
95 connectionEventCallback_->onConnectionDequeuedByAcceptCallback(
98 callback_->connectionAccepted(msg.fd, msg.address);
101 case MessageType::MSG_ERROR:
103 std::runtime_error ex(msg.msg);
104 callback_->acceptError(ex);
109 LOG(ERROR) << "invalid accept notification message type "
111 std::runtime_error ex(
112 "received invalid accept notification message type");
113 callback_->acceptError(ex);
119 * AsyncServerSocket::BackoffTimeout
121 class AsyncServerSocket::BackoffTimeout : public AsyncTimeout {
123 // Disallow copy, move, and default constructors.
124 BackoffTimeout(BackoffTimeout&&) = delete;
125 BackoffTimeout(AsyncServerSocket* socket)
126 : AsyncTimeout(socket->getEventBase()), socket_(socket) {}
128 void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); }
131 AsyncServerSocket* socket_;
135 * AsyncServerSocket methods
138 AsyncServerSocket::AsyncServerSocket(EventBase* eventBase)
139 : eventBase_(eventBase),
141 maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce),
142 maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue),
143 acceptRateAdjustSpeed_(0),
145 lastAccepTimestamp_(std::chrono::steady_clock::now()),
146 numDroppedConnections_(0),
148 backoffTimeout_(nullptr),
150 keepAliveEnabled_(true),
152 shutdownSocketSet_(nullptr) {
155 void AsyncServerSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
156 if (shutdownSocketSet_ == newSS) {
159 if (shutdownSocketSet_) {
160 for (auto& h : sockets_) {
161 shutdownSocketSet_->remove(h.socket_);
164 shutdownSocketSet_ = newSS;
165 if (shutdownSocketSet_) {
166 for (auto& h : sockets_) {
167 shutdownSocketSet_->add(h.socket_);
172 AsyncServerSocket::~AsyncServerSocket() {
173 assert(callbacks_.empty());
176 int AsyncServerSocket::stopAccepting(int shutdownFlags) {
178 for (auto& handler : sockets_) {
179 VLOG(10) << "AsyncServerSocket::stopAccepting " << this <<
182 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
184 // When destroy is called, unregister and close the socket immediately
187 for (auto& handler : sockets_) {
188 handler.unregisterHandler();
189 if (shutdownSocketSet_) {
190 shutdownSocketSet_->close(handler.socket_);
191 } else if (shutdownFlags >= 0) {
192 result = shutdownNoInt(handler.socket_, shutdownFlags);
193 pendingCloseSockets_.push_back(handler.socket_);
195 closeNoInt(handler.socket_);
200 // Destroy the backoff timout. This will cancel it if it is running.
201 delete backoffTimeout_;
202 backoffTimeout_ = nullptr;
204 // Close all of the callback queues to notify them that they are being
205 // destroyed. No one should access the AsyncServerSocket any more once
206 // destroy() is called. However, clear out callbacks_ before invoking the
207 // accept callbacks just in case. This will potentially help us detect the
208 // bug if one of the callbacks calls addAcceptCallback() or
209 // removeAcceptCallback().
210 std::vector<CallbackInfo> callbacksCopy;
211 callbacks_.swap(callbacksCopy);
212 for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
213 it != callbacksCopy.end();
215 it->consumer->stop(it->eventBase, it->callback);
221 void AsyncServerSocket::destroy() {
223 for (auto s : pendingCloseSockets_) {
226 // Then call DelayedDestruction::destroy() to take care of
227 // whether or not we need immediate or delayed destruction
228 DelayedDestruction::destroy();
231 void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
232 assert(eventBase_ == nullptr);
233 assert(eventBase->isInEventBaseThread());
235 eventBase_ = eventBase;
236 for (auto& handler : sockets_) {
237 handler.attachEventBase(eventBase);
241 void AsyncServerSocket::detachEventBase() {
242 assert(eventBase_ != nullptr);
243 assert(eventBase_->isInEventBaseThread());
246 eventBase_ = nullptr;
247 for (auto& handler : sockets_) {
248 handler.detachEventBase();
252 void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
253 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
255 if (sockets_.size() > 0) {
256 throw std::invalid_argument(
257 "cannot call useExistingSocket() on a "
258 "AsyncServerSocket that already has a socket");
262 // Set addressFamily_ from this socket.
263 // Note that the socket may not have been bound yet, but
264 // setFromLocalAddress() will still work and get the correct address family.
265 // We will update addressFamily_ again anyway if bind() is called later.
266 SocketAddress address;
267 address.setFromLocalAddress(fd);
270 sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
271 sockets_.back().changeHandlerFD(fd);
275 void AsyncServerSocket::useExistingSocket(int fd) {
276 useExistingSockets({fd});
279 void AsyncServerSocket::bindSocket(
281 const SocketAddress& address,
282 bool isExistingSocket) {
283 sockaddr_storage addrStorage;
284 address.getAddress(&addrStorage);
285 sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
286 if (::bind(fd, saddr, address.getActualSize()) != 0) {
287 if (!isExistingSocket) {
290 folly::throwSystemError(errno,
291 "failed to bind to async server socket: " +
295 // If we just created this socket, update the EventHandler and set socket_
296 if (!isExistingSocket) {
297 sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
301 void AsyncServerSocket::bind(const SocketAddress& address) {
302 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
304 // useExistingSocket() may have been called to initialize socket_ already.
305 // However, in the normal case we need to create a new socket now.
306 // Don't set socket_ yet, so that socket_ will remain uninitialized if an
309 if (sockets_.size() == 0) {
310 fd = createSocket(address.getFamily());
311 } else if (sockets_.size() == 1) {
312 if (address.getFamily() != sockets_[0].addressFamily_) {
313 throw std::invalid_argument(
314 "Attempted to bind address to socket with "
315 "different address family");
317 fd = sockets_[0].socket_;
319 throw std::invalid_argument(
320 "Attempted to bind to multiple fds");
323 bindSocket(fd, address, !sockets_.empty());
326 void AsyncServerSocket::bind(
327 const std::vector<IPAddress>& ipAddresses,
329 if (ipAddresses.empty()) {
330 throw std::invalid_argument("No ip addresses were provided");
332 if (!sockets_.empty()) {
333 throw std::invalid_argument("Cannot call bind on a AsyncServerSocket "
334 "that already has a socket.");
337 for (const IPAddress& ipAddress : ipAddresses) {
338 SocketAddress address(ipAddress.toFullyQualified(), port);
339 int fd = createSocket(address.getFamily());
341 bindSocket(fd, address, false);
343 if (sockets_.size() == 0) {
344 throw std::runtime_error(
345 "did not bind any async server socket for port and addresses");
349 void AsyncServerSocket::bind(uint16_t port) {
350 struct addrinfo hints, *res, *res0;
351 char sport[sizeof("65536")];
353 memset(&hints, 0, sizeof(hints));
354 hints.ai_family = AF_UNSPEC;
355 hints.ai_socktype = SOCK_STREAM;
356 hints.ai_flags = AI_PASSIVE;
357 snprintf(sport, sizeof(sport), "%u", port);
359 if (getaddrinfo(nullptr, sport, &hints, &res0)) {
360 throw std::invalid_argument(
361 "Attempted to bind address to socket with "
365 SCOPE_EXIT { freeaddrinfo(res0); };
367 auto setupAddress = [&] (struct addrinfo* res) {
368 int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
369 // IPv6/IPv4 may not be supported by the kernel
370 if (s < 0 && errno == EAFNOSUPPORT) {
382 if (res->ai_family == AF_INET6) {
384 CHECK(0 == setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY,
385 &v6only, sizeof(v6only)));
388 SocketAddress address;
389 address.setFromLocalAddress(s);
391 sockets_.emplace_back(eventBase_, s, this, address.getFamily());
393 // Bind to the socket
394 if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) {
395 folly::throwSystemError(
397 "failed to bind to async server socket for port");
401 const int kNumTries = 25;
402 for (int tries = 1; true; tries++) {
403 // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
404 // should return IPv6 first and then IPv4 addresses, but glibc's
405 // getaddrinfo(nullptr) with AI_PASSIVE returns:
406 // - 0.0.0.0 (IPv4-only)
407 // - :: (IPv6+IPv4) in this order
408 // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
409 for (res = res0; res; res = res->ai_next) {
410 if (res->ai_family == AF_INET6) {
415 // If port == 0, then we should try to bind to the same port on ipv4 and
416 // ipv6. So if we did bind to ipv6, figure out that port and use it,
417 // except for the last attempt when we just use any port available.
418 if (sockets_.size() == 1 && port == 0) {
419 SocketAddress address;
420 address.setFromLocalAddress(sockets_.back().socket_);
421 snprintf(sport, sizeof(sport), "%u", address.getPort());
423 CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
427 for (res = res0; res; res = res->ai_next) {
428 if (res->ai_family != AF_INET6) {
432 } catch (const std::system_error& e) {
433 // if we can't bind to the same port on ipv4 as ipv6 when using port=0
434 // then we will try again another 2 times before giving up. We do this
435 // by closing the sockets that were opened, then redoing the whole thing
436 if (port == 0 && !sockets_.empty() && tries != kNumTries) {
437 for (const auto& socket : sockets_) {
438 if (socket.socket_ <= 0) {
440 } else if (shutdownSocketSet_) {
441 shutdownSocketSet_->close(socket.socket_);
443 closeNoInt(socket.socket_);
447 snprintf(sport, sizeof(sport), "%u", port);
449 CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
458 if (sockets_.size() == 0) {
459 throw std::runtime_error(
460 "did not bind any async server socket for port");
464 void AsyncServerSocket::listen(int backlog) {
465 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
468 for (auto& handler : sockets_) {
469 if (::listen(handler.socket_, backlog) == -1) {
470 folly::throwSystemError(errno,
471 "failed to listen on async server socket");
476 void AsyncServerSocket::getAddress(SocketAddress* addressReturn) const {
477 CHECK(sockets_.size() >= 1);
478 VLOG_IF(2, sockets_.size() > 1)
479 << "Warning: getAddress() called and multiple addresses available ("
480 << sockets_.size() << "). Returning only the first one.";
482 addressReturn->setFromLocalAddress(sockets_[0].socket_);
485 std::vector<SocketAddress> AsyncServerSocket::getAddresses()
487 CHECK(sockets_.size() >= 1);
488 auto tsaVec = std::vector<SocketAddress>(sockets_.size());
489 auto tsaIter = tsaVec.begin();
490 for (const auto& socket : sockets_) {
491 (tsaIter++)->setFromLocalAddress(socket.socket_);
496 void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
497 EventBase *eventBase,
498 uint32_t maxAtOnce) {
499 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
501 // If this is the first accept callback and we are supposed to be accepting,
502 // start accepting once the callback is installed.
503 bool runStartAccepting = accepting_ && callbacks_.empty();
506 eventBase = eventBase_; // Run in AsyncServerSocket's eventbase
509 callbacks_.emplace_back(callback, eventBase);
511 // Start the remote acceptor.
513 // It would be nice if we could avoid starting the remote acceptor if
514 // eventBase == eventBase_. However, that would cause issues if
515 // detachEventBase() and attachEventBase() were ever used to change the
516 // primary EventBase for the server socket. Therefore we require the caller
517 // to specify a nullptr EventBase if they want to ensure that the callback is
518 // always invoked in the primary EventBase, and to be able to invoke that
519 // callback more efficiently without having to use a notification queue.
520 RemoteAcceptor* acceptor = nullptr;
522 acceptor = new RemoteAcceptor(callback, connectionEventCallback_);
523 acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
525 callbacks_.pop_back();
529 callbacks_.back().consumer = acceptor;
531 // If this is the first accept callback and we are supposed to be accepting,
533 if (runStartAccepting) {
538 void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
539 EventBase *eventBase) {
540 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
542 // Find the matching AcceptCallback.
543 // We just do a simple linear search; we don't expect removeAcceptCallback()
544 // to be called frequently, and we expect there to only be a small number of
546 std::vector<CallbackInfo>::iterator it = callbacks_.begin();
549 if (it == callbacks_.end()) {
550 throw std::runtime_error("AsyncServerSocket::removeAcceptCallback(): "
551 "accept callback not found");
553 if (it->callback == callback &&
554 (it->eventBase == eventBase || eventBase == nullptr)) {
561 // Remove this callback from callbacks_.
563 // Do this before invoking the acceptStopped() callback, in case
564 // acceptStopped() invokes one of our methods that examines callbacks_.
566 // Save a copy of the CallbackInfo first.
567 CallbackInfo info(*it);
568 callbacks_.erase(it);
569 if (n < callbackIndex_) {
570 // We removed an element before callbackIndex_. Move callbackIndex_ back
571 // one step, since things after n have been shifted back by 1.
574 // We removed something at or after callbackIndex_.
575 // If we removed the last element and callbackIndex_ was pointing at it,
576 // we need to reset callbackIndex_ to 0.
577 if (callbackIndex_ >= callbacks_.size()) {
582 info.consumer->stop(info.eventBase, info.callback);
584 // If we are supposed to be accepting but the last accept callback
585 // was removed, unregister for events until a callback is added.
586 if (accepting_ && callbacks_.empty()) {
587 for (auto& handler : sockets_) {
588 handler.unregisterHandler();
593 void AsyncServerSocket::startAccepting() {
594 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
597 if (callbacks_.empty()) {
598 // We can't actually begin accepting if no callbacks are defined.
599 // Wait until a callback is added to start accepting.
603 for (auto& handler : sockets_) {
604 if (!handler.registerHandler(
605 EventHandler::READ | EventHandler::PERSIST)) {
606 throw std::runtime_error("failed to register for accept events");
611 void AsyncServerSocket::pauseAccepting() {
612 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
614 for (auto& handler : sockets_) {
615 handler. unregisterHandler();
618 // If we were in the accept backoff state, disable the backoff timeout
619 if (backoffTimeout_) {
620 backoffTimeout_->cancelTimeout();
624 int AsyncServerSocket::createSocket(int family) {
625 int fd = socket(family, SOCK_STREAM, 0);
627 folly::throwSystemError(errno, "error creating async server socket");
639 void AsyncServerSocket::setupSocket(int fd) {
640 // Get the address family
641 SocketAddress address;
642 address.setFromLocalAddress(fd);
644 // Put the socket in non-blocking mode
645 if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
646 folly::throwSystemError(errno,
647 "failed to put socket in non-blocking mode");
650 // Set reuseaddr to avoid 2MSL delay on server restart
652 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
653 // This isn't a fatal error; just log an error message and continue
654 LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
657 // Set reuseport to support multiple accept threads
659 if (reusePortEnabled_ &&
660 setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
661 LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
663 folly::throwSystemError(errno,
664 "failed to bind to async server socket: " +
668 // Set keepalive as desired
669 if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
670 (keepAliveEnabled_) ? &one : &zero, sizeof(int)) != 0) {
671 LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: " <<
675 // Setup FD_CLOEXEC flag
677 (-1 == folly::setCloseOnExec(fd, closeOnExec_))) {
678 LOG(ERROR) << "failed to set FD_CLOEXEC on async server socket: " <<
682 // Set TCP nodelay if available, MAC OS X Hack
683 // See http://lists.danga.com/pipermail/memcached/2005-March/001240.html
685 auto family = address.getFamily();
686 if (family != AF_UNIX) {
687 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0) {
688 // This isn't a fatal error; just log an error message and continue
689 LOG(ERROR) << "failed to set TCP_NODELAY on async server socket: " <<
695 if (shutdownSocketSet_) {
696 shutdownSocketSet_->add(fd);
700 void AsyncServerSocket::handlerReady(
701 uint16_t events, int fd, sa_family_t addressFamily) noexcept {
702 assert(!callbacks_.empty());
703 DestructorGuard dg(this);
705 // Only accept up to maxAcceptAtOnce_ connections at a time,
706 // to avoid starving other I/O handlers using this EventBase.
707 for (uint32_t n = 0; n < maxAcceptAtOnce_; ++n) {
708 SocketAddress address;
710 sockaddr_storage addrStorage;
711 socklen_t addrLen = sizeof(addrStorage);
712 sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
714 // In some cases, accept() doesn't seem to update these correctly.
715 saddr->sa_family = addressFamily;
716 if (addressFamily == AF_UNIX) {
717 addrLen = sizeof(struct sockaddr_un);
720 // Accept a new client socket
722 int clientSocket = accept4(fd, saddr, &addrLen, SOCK_NONBLOCK);
724 int clientSocket = accept(fd, saddr, &addrLen);
727 address.setFromSockaddr(saddr, addrLen);
729 if (clientSocket >= 0 && connectionEventCallback_) {
730 connectionEventCallback_->onConnectionAccepted(clientSocket, address);
733 std::chrono::time_point<std::chrono::steady_clock> nowMs =
734 std::chrono::steady_clock::now();
735 auto timeSinceLastAccept = std::max<int64_t>(
737 nowMs.time_since_epoch().count() -
738 lastAccepTimestamp_.time_since_epoch().count());
739 lastAccepTimestamp_ = nowMs;
740 if (acceptRate_ < 1) {
741 acceptRate_ *= 1 + acceptRateAdjustSpeed_ * timeSinceLastAccept;
742 if (acceptRate_ >= 1) {
744 } else if (rand() > acceptRate_ * RAND_MAX) {
745 ++numDroppedConnections_;
746 if (clientSocket >= 0) {
747 closeNoInt(clientSocket);
748 if (connectionEventCallback_) {
749 connectionEventCallback_->onConnectionDropped(clientSocket,
757 if (clientSocket < 0) {
758 if (errno == EAGAIN) {
759 // No more sockets to accept right now.
760 // Check for this code first, since it's the most common.
762 } else if (errno == EMFILE || errno == ENFILE) {
763 // We're out of file descriptors. Perhaps we're accepting connections
764 // too quickly. Pause accepting briefly to back off and give the server
765 // a chance to recover.
766 LOG(ERROR) << "accept failed: out of file descriptors; entering accept "
770 // Dispatch the error message
771 dispatchError("accept() failed", errno);
773 dispatchError("accept() failed", errno);
775 if (connectionEventCallback_) {
776 connectionEventCallback_->onConnectionAcceptError(errno);
781 #ifndef SOCK_NONBLOCK
782 // Explicitly set the new connection to non-blocking mode
783 if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
784 closeNoInt(clientSocket);
785 dispatchError("failed to set accepted socket to non-blocking mode",
787 if (connectionEventCallback_) {
788 connectionEventCallback_->onConnectionDropped(clientSocket, address);
794 // Inform the callback about the new connection
795 dispatchSocket(clientSocket, std::move(address));
797 // If we aren't accepting any more, break out of the loop
798 if (!accepting_ || callbacks_.empty()) {
804 void AsyncServerSocket::dispatchSocket(int socket,
805 SocketAddress&& address) {
806 uint32_t startingIndex = callbackIndex_;
808 // Short circuit if the callback is in the primary EventBase thread
810 CallbackInfo *info = nextCallback();
811 if (info->eventBase == nullptr) {
812 info->callback->connectionAccepted(socket, address);
816 const SocketAddress addr(address);
817 // Create a message to send over the notification queue
819 msg.type = MessageType::MSG_NEW_CONN;
820 msg.address = std::move(address);
823 // Loop until we find a free queue to write to
825 if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
826 if (connectionEventCallback_) {
827 connectionEventCallback_->onConnectionEnqueuedForAcceptCallback(socket,
834 // We couldn't add to queue. Fall through to below
836 ++numDroppedConnections_;
837 if (acceptRateAdjustSpeed_ > 0) {
838 // aggressively decrease accept rate when in trouble
839 static const double kAcceptRateDecreaseSpeed = 0.1;
840 acceptRate_ *= 1 - kAcceptRateDecreaseSpeed;
844 if (callbackIndex_ == startingIndex) {
845 // The notification queue was full
846 // We can't really do anything at this point other than close the socket.
848 // This should only happen if a user's service is behaving extremely
849 // badly and none of the EventBase threads are looping fast enough to
850 // process the incoming connections. If the service is overloaded, it
851 // should use pauseAccepting() to temporarily back off accepting new
852 // connections, before they reach the point where their threads can't
853 // even accept new messages.
854 LOG(ERROR) << "failed to dispatch newly accepted socket:"
855 << " all accept callback queues are full";
857 if (connectionEventCallback_) {
858 connectionEventCallback_->onConnectionDropped(socket, addr);
863 info = nextCallback();
867 void AsyncServerSocket::dispatchError(const char *msgstr, int errnoValue) {
868 uint32_t startingIndex = callbackIndex_;
869 CallbackInfo *info = nextCallback();
871 // Create a message to send over the notification queue
873 msg.type = MessageType::MSG_ERROR;
874 msg.err = errnoValue;
875 msg.msg = std::move(msgstr);
878 // Short circuit if the callback is in the primary EventBase thread
879 if (info->eventBase == nullptr) {
880 std::runtime_error ex(
881 std::string(msgstr) + folly::to<std::string>(errnoValue));
882 info->callback->acceptError(ex);
886 if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
889 // Fall through and try another callback
891 if (callbackIndex_ == startingIndex) {
892 // The notification queues for all of the callbacks were full.
893 // We can't really do anything at this point.
894 LOG(ERROR) << "failed to dispatch accept error: all accept callback "
895 "queues are full: error msg: " <<
896 msg.msg.c_str() << errnoValue;
899 info = nextCallback();
903 void AsyncServerSocket::enterBackoff() {
904 // If this is the first time we have entered the backoff state,
905 // allocate backoffTimeout_.
906 if (backoffTimeout_ == nullptr) {
908 backoffTimeout_ = new BackoffTimeout(this);
909 } catch (const std::bad_alloc& ex) {
910 // Man, we couldn't even allocate the timer to re-enable accepts.
911 // We must be in pretty bad shape. Don't pause accepting for now,
912 // since we won't be able to re-enable ourselves later.
913 LOG(ERROR) << "failed to allocate AsyncServerSocket backoff"
914 << " timer; unable to temporarly pause accepting";
915 if (connectionEventCallback_) {
916 connectionEventCallback_->onBackoffError();
922 // For now, we simply pause accepting for 1 second.
924 // We could add some smarter backoff calculation here in the future. (e.g.,
925 // start sleeping for longer if we keep hitting the backoff frequently.)
926 // Typically the user needs to figure out why the server is overloaded and
927 // fix it in some other way, though. The backoff timer is just a simple
928 // mechanism to try and give the connection processing code a little bit of
929 // breathing room to catch up, and to avoid just spinning and failing to
930 // accept over and over again.
931 const uint32_t timeoutMS = 1000;
932 if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
933 LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
934 << "unable to temporarly pause accepting";
935 if (connectionEventCallback_) {
936 connectionEventCallback_->onBackoffError();
941 // The backoff timer is scheduled to re-enable accepts.
942 // Go ahead and disable accepts for now. We leave accepting_ set to true,
943 // since that tracks the desired state requested by the user.
944 for (auto& handler : sockets_) {
945 handler.unregisterHandler();
947 if (connectionEventCallback_) {
948 connectionEventCallback_->onBackoffStarted();
952 void AsyncServerSocket::backoffTimeoutExpired() {
953 // accepting_ should still be true.
954 // If pauseAccepting() was called while in the backoff state it will cancel
955 // the backoff timeout.
957 // We can't be detached from the EventBase without being paused
958 assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread());
960 // If all of the callbacks were removed, we shouldn't re-enable accepts
961 if (callbacks_.empty()) {
962 if (connectionEventCallback_) {
963 connectionEventCallback_->onBackoffEnded();
968 // Register the handler.
969 for (auto& handler : sockets_) {
970 if (!handler.registerHandler(
971 EventHandler::READ | EventHandler::PERSIST)) {
972 // We're hosed. We could just re-schedule backoffTimeout_ to
973 // re-try again after a little bit. However, we don't want to
974 // loop retrying forever if we can't re-enable accepts. Just
975 // abort the entire program in this state; things are really bad
976 // and restarting the entire server is probably the best remedy.
978 << "failed to re-enable AsyncServerSocket accepts after backoff; "
983 if (connectionEventCallback_) {
984 connectionEventCallback_->onBackoffEnded();