2 * Copyright 2016 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/detail/SocketFastOpen.h>
26 #include <folly/io/async/EventBase.h>
27 #include <folly/io/async/NotificationQueue.h>
28 #include <folly/portability/Fcntl.h>
31 #include <netinet/tcp.h>
33 #include <sys/socket.h>
34 #include <sys/types.h>
39 const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce;
40 const uint32_t AsyncServerSocket::kDefaultCallbackAcceptAtOnce;
41 const uint32_t AsyncServerSocket::kDefaultMaxMessagesInQueue;
43 int setCloseOnExec(int fd, int value) {
44 // Read the current flags
45 int old_flags = fcntl(fd, F_GETFD, 0);
47 // If reading the flags failed, return error indication now
51 // Set just the flag we want to set
54 new_flags = old_flags | FD_CLOEXEC;
56 new_flags = old_flags & ~FD_CLOEXEC;
58 // Store modified flag word in the descriptor
59 return fcntl(fd, F_SETFD, new_flags);
62 void AsyncServerSocket::RemoteAcceptor::start(
63 EventBase* eventBase, uint32_t maxAtOnce, uint32_t maxInQueue) {
64 setMaxReadAtOnce(maxAtOnce);
65 queue_.setMaxQueueSize(maxInQueue);
67 if (!eventBase->runInEventBaseThread([=](){
68 callback_->acceptStarted();
69 this->startConsuming(eventBase, &queue_);
71 throw std::invalid_argument("unable to start waiting on accept "
72 "notification queue in the specified "
77 void AsyncServerSocket::RemoteAcceptor::stop(
78 EventBase* eventBase, AcceptCallback* callback) {
79 if (!eventBase->runInEventBaseThread([=](){
80 callback->acceptStopped();
83 throw std::invalid_argument("unable to start waiting on accept "
84 "notification queue in the specified "
89 void AsyncServerSocket::RemoteAcceptor::messageAvailable(
93 case MessageType::MSG_NEW_CONN:
95 if (connectionEventCallback_) {
96 connectionEventCallback_->onConnectionDequeuedByAcceptorCallback(
99 callback_->connectionAccepted(msg.fd, msg.address);
102 case MessageType::MSG_ERROR:
104 std::runtime_error ex(msg.msg);
105 callback_->acceptError(ex);
110 LOG(ERROR) << "invalid accept notification message type "
112 std::runtime_error ex(
113 "received invalid accept notification message type");
114 callback_->acceptError(ex);
120 * AsyncServerSocket::BackoffTimeout
122 class AsyncServerSocket::BackoffTimeout : public AsyncTimeout {
124 // Disallow copy, move, and default constructors.
125 BackoffTimeout(BackoffTimeout&&) = delete;
126 BackoffTimeout(AsyncServerSocket* socket)
127 : AsyncTimeout(socket->getEventBase()), socket_(socket) {}
129 void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); }
132 AsyncServerSocket* socket_;
136 * AsyncServerSocket methods
139 AsyncServerSocket::AsyncServerSocket(EventBase* eventBase)
140 : eventBase_(eventBase),
142 maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce),
143 maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue),
144 acceptRateAdjustSpeed_(0),
146 lastAccepTimestamp_(std::chrono::steady_clock::now()),
147 numDroppedConnections_(0),
149 backoffTimeout_(nullptr),
151 keepAliveEnabled_(true),
153 shutdownSocketSet_(nullptr) {
156 void AsyncServerSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
157 if (shutdownSocketSet_ == newSS) {
160 if (shutdownSocketSet_) {
161 for (auto& h : sockets_) {
162 shutdownSocketSet_->remove(h.socket_);
165 shutdownSocketSet_ = newSS;
166 if (shutdownSocketSet_) {
167 for (auto& h : sockets_) {
168 shutdownSocketSet_->add(h.socket_);
173 AsyncServerSocket::~AsyncServerSocket() {
174 assert(callbacks_.empty());
177 int AsyncServerSocket::stopAccepting(int shutdownFlags) {
179 for (auto& handler : sockets_) {
180 VLOG(10) << "AsyncServerSocket::stopAccepting " << this <<
183 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
185 // When destroy is called, unregister and close the socket immediately.
188 // Close the sockets in reverse order as they were opened to avoid
189 // the condition where another process concurrently tries to open
190 // the same port, succeed to bind the first socket but fails on the
191 // second because it hasn't been closed yet.
192 for (; !sockets_.empty(); sockets_.pop_back()) {
193 auto& handler = sockets_.back();
194 handler.unregisterHandler();
195 if (shutdownSocketSet_) {
196 shutdownSocketSet_->close(handler.socket_);
197 } else if (shutdownFlags >= 0) {
198 result = shutdownNoInt(handler.socket_, shutdownFlags);
199 pendingCloseSockets_.push_back(handler.socket_);
201 closeNoInt(handler.socket_);
205 // Destroy the backoff timout. This will cancel it if it is running.
206 delete backoffTimeout_;
207 backoffTimeout_ = nullptr;
209 // Close all of the callback queues to notify them that they are being
210 // destroyed. No one should access the AsyncServerSocket any more once
211 // destroy() is called. However, clear out callbacks_ before invoking the
212 // accept callbacks just in case. This will potentially help us detect the
213 // bug if one of the callbacks calls addAcceptCallback() or
214 // removeAcceptCallback().
215 std::vector<CallbackInfo> callbacksCopy;
216 callbacks_.swap(callbacksCopy);
217 for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
218 it != callbacksCopy.end();
220 it->consumer->stop(it->eventBase, it->callback);
226 void AsyncServerSocket::destroy() {
228 for (auto s : pendingCloseSockets_) {
231 // Then call DelayedDestruction::destroy() to take care of
232 // whether or not we need immediate or delayed destruction
233 DelayedDestruction::destroy();
236 void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
237 assert(eventBase_ == nullptr);
238 assert(eventBase->isInEventBaseThread());
240 eventBase_ = eventBase;
241 for (auto& handler : sockets_) {
242 handler.attachEventBase(eventBase);
246 void AsyncServerSocket::detachEventBase() {
247 assert(eventBase_ != nullptr);
248 assert(eventBase_->isInEventBaseThread());
251 eventBase_ = nullptr;
252 for (auto& handler : sockets_) {
253 handler.detachEventBase();
257 void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
258 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
260 if (sockets_.size() > 0) {
261 throw std::invalid_argument(
262 "cannot call useExistingSocket() on a "
263 "AsyncServerSocket that already has a socket");
267 // Set addressFamily_ from this socket.
268 // Note that the socket may not have been bound yet, but
269 // setFromLocalAddress() will still work and get the correct address family.
270 // We will update addressFamily_ again anyway if bind() is called later.
271 SocketAddress address;
272 address.setFromLocalAddress(fd);
274 setupSocket(fd, address.getFamily());
275 sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
276 sockets_.back().changeHandlerFD(fd);
280 void AsyncServerSocket::useExistingSocket(int fd) {
281 useExistingSockets({fd});
284 void AsyncServerSocket::bindSocket(
286 const SocketAddress& address,
287 bool isExistingSocket) {
288 sockaddr_storage addrStorage;
289 address.getAddress(&addrStorage);
290 sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
291 if (::bind(fd, saddr, address.getActualSize()) != 0) {
292 if (!isExistingSocket) {
295 folly::throwSystemError(errno,
296 "failed to bind to async server socket: " +
300 // If we just created this socket, update the EventHandler and set socket_
301 if (!isExistingSocket) {
302 sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
306 void AsyncServerSocket::bind(const SocketAddress& address) {
307 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
309 // useExistingSocket() may have been called to initialize socket_ already.
310 // However, in the normal case we need to create a new socket now.
311 // Don't set socket_ yet, so that socket_ will remain uninitialized if an
314 if (sockets_.size() == 0) {
315 fd = createSocket(address.getFamily());
316 } else if (sockets_.size() == 1) {
317 if (address.getFamily() != sockets_[0].addressFamily_) {
318 throw std::invalid_argument(
319 "Attempted to bind address to socket with "
320 "different address family");
322 fd = sockets_[0].socket_;
324 throw std::invalid_argument(
325 "Attempted to bind to multiple fds");
328 bindSocket(fd, address, !sockets_.empty());
331 void AsyncServerSocket::bind(
332 const std::vector<IPAddress>& ipAddresses,
334 if (ipAddresses.empty()) {
335 throw std::invalid_argument("No ip addresses were provided");
337 if (!sockets_.empty()) {
338 throw std::invalid_argument("Cannot call bind on a AsyncServerSocket "
339 "that already has a socket.");
342 for (const IPAddress& ipAddress : ipAddresses) {
343 SocketAddress address(ipAddress.toFullyQualified(), port);
344 int fd = createSocket(address.getFamily());
346 bindSocket(fd, address, false);
348 if (sockets_.size() == 0) {
349 throw std::runtime_error(
350 "did not bind any async server socket for port and addresses");
354 void AsyncServerSocket::bind(uint16_t port) {
355 struct addrinfo hints, *res, *res0;
356 char sport[sizeof("65536")];
358 memset(&hints, 0, sizeof(hints));
359 hints.ai_family = AF_UNSPEC;
360 hints.ai_socktype = SOCK_STREAM;
361 hints.ai_flags = AI_PASSIVE;
362 snprintf(sport, sizeof(sport), "%u", port);
364 if (getaddrinfo(nullptr, sport, &hints, &res0)) {
365 throw std::invalid_argument(
366 "Attempted to bind address to socket with "
370 SCOPE_EXIT { freeaddrinfo(res0); };
372 auto setupAddress = [&] (struct addrinfo* res) {
373 int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
374 // IPv6/IPv4 may not be supported by the kernel
375 if (s < 0 && errno == EAFNOSUPPORT) {
381 setupSocket(s, res->ai_family);
387 if (res->ai_family == AF_INET6) {
389 CHECK(0 == setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY,
390 &v6only, sizeof(v6only)));
393 SocketAddress address;
394 address.setFromLocalAddress(s);
396 sockets_.emplace_back(eventBase_, s, this, address.getFamily());
398 // Bind to the socket
399 if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) {
400 folly::throwSystemError(
402 "failed to bind to async server socket for port ",
403 SocketAddress::getPortFrom(res->ai_addr),
405 SocketAddress::getFamilyNameFrom(res->ai_addr, "<unknown>"));
409 const int kNumTries = 25;
410 for (int tries = 1; true; tries++) {
411 // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
412 // should return IPv6 first and then IPv4 addresses, but glibc's
413 // getaddrinfo(nullptr) with AI_PASSIVE returns:
414 // - 0.0.0.0 (IPv4-only)
415 // - :: (IPv6+IPv4) in this order
416 // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
417 for (res = res0; res; res = res->ai_next) {
418 if (res->ai_family == AF_INET6) {
423 // If port == 0, then we should try to bind to the same port on ipv4 and
424 // ipv6. So if we did bind to ipv6, figure out that port and use it.
425 if (sockets_.size() == 1 && port == 0) {
426 SocketAddress address;
427 address.setFromLocalAddress(sockets_.back().socket_);
428 snprintf(sport, sizeof(sport), "%u", address.getPort());
430 CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
434 for (res = res0; res; res = res->ai_next) {
435 if (res->ai_family != AF_INET6) {
439 } catch (const std::system_error& e) {
440 // If we can't bind to the same port on ipv4 as ipv6 when using
441 // port=0 then we will retry again before giving up after
442 // kNumTries attempts. We do this by closing the sockets that
443 // were opened, then restarting from scratch.
444 if (port == 0 && !sockets_.empty() && tries != kNumTries) {
445 for (const auto& socket : sockets_) {
446 if (socket.socket_ <= 0) {
448 } else if (shutdownSocketSet_) {
449 shutdownSocketSet_->close(socket.socket_);
451 closeNoInt(socket.socket_);
455 snprintf(sport, sizeof(sport), "%u", port);
457 CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
467 if (sockets_.size() == 0) {
468 throw std::runtime_error(
469 "did not bind any async server socket for port");
473 void AsyncServerSocket::listen(int backlog) {
474 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
477 for (auto& handler : sockets_) {
478 if (::listen(handler.socket_, backlog) == -1) {
479 folly::throwSystemError(errno,
480 "failed to listen on async server socket");
485 void AsyncServerSocket::getAddress(SocketAddress* addressReturn) const {
486 CHECK(sockets_.size() >= 1);
487 VLOG_IF(2, sockets_.size() > 1)
488 << "Warning: getAddress() called and multiple addresses available ("
489 << sockets_.size() << "). Returning only the first one.";
491 addressReturn->setFromLocalAddress(sockets_[0].socket_);
494 std::vector<SocketAddress> AsyncServerSocket::getAddresses()
496 CHECK(sockets_.size() >= 1);
497 auto tsaVec = std::vector<SocketAddress>(sockets_.size());
498 auto tsaIter = tsaVec.begin();
499 for (const auto& socket : sockets_) {
500 (tsaIter++)->setFromLocalAddress(socket.socket_);
505 void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
506 EventBase *eventBase,
507 uint32_t maxAtOnce) {
508 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
510 // If this is the first accept callback and we are supposed to be accepting,
511 // start accepting once the callback is installed.
512 bool runStartAccepting = accepting_ && callbacks_.empty();
515 eventBase = eventBase_; // Run in AsyncServerSocket's eventbase
518 callbacks_.emplace_back(callback, eventBase);
520 // Start the remote acceptor.
522 // It would be nice if we could avoid starting the remote acceptor if
523 // eventBase == eventBase_. However, that would cause issues if
524 // detachEventBase() and attachEventBase() were ever used to change the
525 // primary EventBase for the server socket. Therefore we require the caller
526 // to specify a nullptr EventBase if they want to ensure that the callback is
527 // always invoked in the primary EventBase, and to be able to invoke that
528 // callback more efficiently without having to use a notification queue.
529 RemoteAcceptor* acceptor = nullptr;
531 acceptor = new RemoteAcceptor(callback, connectionEventCallback_);
532 acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
534 callbacks_.pop_back();
538 callbacks_.back().consumer = acceptor;
540 // If this is the first accept callback and we are supposed to be accepting,
542 if (runStartAccepting) {
547 void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
548 EventBase *eventBase) {
549 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
551 // Find the matching AcceptCallback.
552 // We just do a simple linear search; we don't expect removeAcceptCallback()
553 // to be called frequently, and we expect there to only be a small number of
555 std::vector<CallbackInfo>::iterator it = callbacks_.begin();
558 if (it == callbacks_.end()) {
559 throw std::runtime_error("AsyncServerSocket::removeAcceptCallback(): "
560 "accept callback not found");
562 if (it->callback == callback &&
563 (it->eventBase == eventBase || eventBase == nullptr)) {
570 // Remove this callback from callbacks_.
572 // Do this before invoking the acceptStopped() callback, in case
573 // acceptStopped() invokes one of our methods that examines callbacks_.
575 // Save a copy of the CallbackInfo first.
576 CallbackInfo info(*it);
577 callbacks_.erase(it);
578 if (n < callbackIndex_) {
579 // We removed an element before callbackIndex_. Move callbackIndex_ back
580 // one step, since things after n have been shifted back by 1.
583 // We removed something at or after callbackIndex_.
584 // If we removed the last element and callbackIndex_ was pointing at it,
585 // we need to reset callbackIndex_ to 0.
586 if (callbackIndex_ >= callbacks_.size()) {
591 info.consumer->stop(info.eventBase, info.callback);
593 // If we are supposed to be accepting but the last accept callback
594 // was removed, unregister for events until a callback is added.
595 if (accepting_ && callbacks_.empty()) {
596 for (auto& handler : sockets_) {
597 handler.unregisterHandler();
602 void AsyncServerSocket::startAccepting() {
603 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
606 if (callbacks_.empty()) {
607 // We can't actually begin accepting if no callbacks are defined.
608 // Wait until a callback is added to start accepting.
612 for (auto& handler : sockets_) {
613 if (!handler.registerHandler(
614 EventHandler::READ | EventHandler::PERSIST)) {
615 throw std::runtime_error("failed to register for accept events");
620 void AsyncServerSocket::pauseAccepting() {
621 assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
623 for (auto& handler : sockets_) {
624 handler. unregisterHandler();
627 // If we were in the accept backoff state, disable the backoff timeout
628 if (backoffTimeout_) {
629 backoffTimeout_->cancelTimeout();
633 int AsyncServerSocket::createSocket(int family) {
634 int fd = socket(family, SOCK_STREAM, 0);
636 folly::throwSystemError(errno, "error creating async server socket");
640 setupSocket(fd, family);
648 void AsyncServerSocket::setupSocket(int fd, int family) {
649 // Put the socket in non-blocking mode
650 if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
651 folly::throwSystemError(errno,
652 "failed to put socket in non-blocking mode");
655 // Set reuseaddr to avoid 2MSL delay on server restart
657 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
658 // This isn't a fatal error; just log an error message and continue
659 LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
662 // Set reuseport to support multiple accept threads
664 if (reusePortEnabled_ &&
665 setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
666 LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
669 folly::throwSystemError(errno, "failed to bind to the async server socket");
671 SocketAddress address;
672 address.setFromLocalAddress(fd);
673 folly::throwSystemError(errno,
674 "failed to bind to async server socket: " +
679 // Set keepalive as desired
680 if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
681 (keepAliveEnabled_) ? &one : &zero, sizeof(int)) != 0) {
682 LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: " <<
686 // Setup FD_CLOEXEC flag
688 (-1 == folly::setCloseOnExec(fd, closeOnExec_))) {
689 LOG(ERROR) << "failed to set FD_CLOEXEC on async server socket: " <<
693 // Set TCP nodelay if available, MAC OS X Hack
694 // See http://lists.danga.com/pipermail/memcached/2005-March/001240.html
696 if (family != AF_UNIX) {
697 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0) {
698 // This isn't a fatal error; just log an error message and continue
699 LOG(ERROR) << "failed to set TCP_NODELAY on async server socket: " <<
706 if (tfo_ && detail::tfo_enable(fd, tfoMaxQueueSize_) != 0) {
707 // This isn't a fatal error; just log an error message and continue
708 LOG(WARNING) << "failed to set TCP_FASTOPEN on async server socket: "
709 << folly::errnoStr(errno);
713 if (shutdownSocketSet_) {
714 shutdownSocketSet_->add(fd);
718 void AsyncServerSocket::handlerReady(uint16_t /* events */,
720 sa_family_t addressFamily) noexcept {
721 assert(!callbacks_.empty());
722 DestructorGuard dg(this);
724 // Only accept up to maxAcceptAtOnce_ connections at a time,
725 // to avoid starving other I/O handlers using this EventBase.
726 for (uint32_t n = 0; n < maxAcceptAtOnce_; ++n) {
727 SocketAddress address;
729 sockaddr_storage addrStorage;
730 socklen_t addrLen = sizeof(addrStorage);
731 sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
733 // In some cases, accept() doesn't seem to update these correctly.
734 saddr->sa_family = addressFamily;
735 if (addressFamily == AF_UNIX) {
736 addrLen = sizeof(struct sockaddr_un);
739 // Accept a new client socket
741 int clientSocket = accept4(fd, saddr, &addrLen, SOCK_NONBLOCK);
743 int clientSocket = accept(fd, saddr, &addrLen);
746 address.setFromSockaddr(saddr, addrLen);
748 if (clientSocket >= 0 && connectionEventCallback_) {
749 connectionEventCallback_->onConnectionAccepted(clientSocket, address);
752 std::chrono::time_point<std::chrono::steady_clock> nowMs =
753 std::chrono::steady_clock::now();
754 auto timeSinceLastAccept = std::max<int64_t>(
756 nowMs.time_since_epoch().count() -
757 lastAccepTimestamp_.time_since_epoch().count());
758 lastAccepTimestamp_ = nowMs;
759 if (acceptRate_ < 1) {
760 acceptRate_ *= 1 + acceptRateAdjustSpeed_ * timeSinceLastAccept;
761 if (acceptRate_ >= 1) {
763 } else if (rand() > acceptRate_ * RAND_MAX) {
764 ++numDroppedConnections_;
765 if (clientSocket >= 0) {
766 closeNoInt(clientSocket);
767 if (connectionEventCallback_) {
768 connectionEventCallback_->onConnectionDropped(clientSocket,
776 if (clientSocket < 0) {
777 if (errno == EAGAIN) {
778 // No more sockets to accept right now.
779 // Check for this code first, since it's the most common.
781 } else if (errno == EMFILE || errno == ENFILE) {
782 // We're out of file descriptors. Perhaps we're accepting connections
783 // too quickly. Pause accepting briefly to back off and give the server
784 // a chance to recover.
785 LOG(ERROR) << "accept failed: out of file descriptors; entering accept "
789 // Dispatch the error message
790 dispatchError("accept() failed", errno);
792 dispatchError("accept() failed", errno);
794 if (connectionEventCallback_) {
795 connectionEventCallback_->onConnectionAcceptError(errno);
800 #ifndef SOCK_NONBLOCK
801 // Explicitly set the new connection to non-blocking mode
802 if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
803 closeNoInt(clientSocket);
804 dispatchError("failed to set accepted socket to non-blocking mode",
806 if (connectionEventCallback_) {
807 connectionEventCallback_->onConnectionDropped(clientSocket, address);
813 // Inform the callback about the new connection
814 dispatchSocket(clientSocket, std::move(address));
816 // If we aren't accepting any more, break out of the loop
817 if (!accepting_ || callbacks_.empty()) {
823 void AsyncServerSocket::dispatchSocket(int socket,
824 SocketAddress&& address) {
825 uint32_t startingIndex = callbackIndex_;
827 // Short circuit if the callback is in the primary EventBase thread
829 CallbackInfo *info = nextCallback();
830 if (info->eventBase == nullptr) {
831 info->callback->connectionAccepted(socket, address);
835 const SocketAddress addr(address);
836 // Create a message to send over the notification queue
838 msg.type = MessageType::MSG_NEW_CONN;
839 msg.address = std::move(address);
842 // Loop until we find a free queue to write to
844 if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
845 if (connectionEventCallback_) {
846 connectionEventCallback_->onConnectionEnqueuedForAcceptorCallback(
854 // We couldn't add to queue. Fall through to below
856 ++numDroppedConnections_;
857 if (acceptRateAdjustSpeed_ > 0) {
858 // aggressively decrease accept rate when in trouble
859 static const double kAcceptRateDecreaseSpeed = 0.1;
860 acceptRate_ *= 1 - kAcceptRateDecreaseSpeed;
864 if (callbackIndex_ == startingIndex) {
865 // The notification queue was full
866 // We can't really do anything at this point other than close the socket.
868 // This should only happen if a user's service is behaving extremely
869 // badly and none of the EventBase threads are looping fast enough to
870 // process the incoming connections. If the service is overloaded, it
871 // should use pauseAccepting() to temporarily back off accepting new
872 // connections, before they reach the point where their threads can't
873 // even accept new messages.
874 LOG(ERROR) << "failed to dispatch newly accepted socket:"
875 << " all accept callback queues are full";
877 if (connectionEventCallback_) {
878 connectionEventCallback_->onConnectionDropped(socket, addr);
883 info = nextCallback();
887 void AsyncServerSocket::dispatchError(const char *msgstr, int errnoValue) {
888 uint32_t startingIndex = callbackIndex_;
889 CallbackInfo *info = nextCallback();
891 // Create a message to send over the notification queue
893 msg.type = MessageType::MSG_ERROR;
894 msg.err = errnoValue;
895 msg.msg = std::move(msgstr);
898 // Short circuit if the callback is in the primary EventBase thread
899 if (info->eventBase == nullptr) {
900 std::runtime_error ex(
901 std::string(msgstr) + folly::to<std::string>(errnoValue));
902 info->callback->acceptError(ex);
906 if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
909 // Fall through and try another callback
911 if (callbackIndex_ == startingIndex) {
912 // The notification queues for all of the callbacks were full.
913 // We can't really do anything at this point.
914 LOG(ERROR) << "failed to dispatch accept error: all accept callback "
915 "queues are full: error msg: " <<
916 msg.msg.c_str() << errnoValue;
919 info = nextCallback();
923 void AsyncServerSocket::enterBackoff() {
924 // If this is the first time we have entered the backoff state,
925 // allocate backoffTimeout_.
926 if (backoffTimeout_ == nullptr) {
928 backoffTimeout_ = new BackoffTimeout(this);
929 } catch (const std::bad_alloc& ex) {
930 // Man, we couldn't even allocate the timer to re-enable accepts.
931 // We must be in pretty bad shape. Don't pause accepting for now,
932 // since we won't be able to re-enable ourselves later.
933 LOG(ERROR) << "failed to allocate AsyncServerSocket backoff"
934 << " timer; unable to temporarly pause accepting";
935 if (connectionEventCallback_) {
936 connectionEventCallback_->onBackoffError();
942 // For now, we simply pause accepting for 1 second.
944 // We could add some smarter backoff calculation here in the future. (e.g.,
945 // start sleeping for longer if we keep hitting the backoff frequently.)
946 // Typically the user needs to figure out why the server is overloaded and
947 // fix it in some other way, though. The backoff timer is just a simple
948 // mechanism to try and give the connection processing code a little bit of
949 // breathing room to catch up, and to avoid just spinning and failing to
950 // accept over and over again.
951 const uint32_t timeoutMS = 1000;
952 if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
953 LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
954 << "unable to temporarly pause accepting";
955 if (connectionEventCallback_) {
956 connectionEventCallback_->onBackoffError();
961 // The backoff timer is scheduled to re-enable accepts.
962 // Go ahead and disable accepts for now. We leave accepting_ set to true,
963 // since that tracks the desired state requested by the user.
964 for (auto& handler : sockets_) {
965 handler.unregisterHandler();
967 if (connectionEventCallback_) {
968 connectionEventCallback_->onBackoffStarted();
972 void AsyncServerSocket::backoffTimeoutExpired() {
973 // accepting_ should still be true.
974 // If pauseAccepting() was called while in the backoff state it will cancel
975 // the backoff timeout.
977 // We can't be detached from the EventBase without being paused
978 assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread());
980 // If all of the callbacks were removed, we shouldn't re-enable accepts
981 if (callbacks_.empty()) {
982 if (connectionEventCallback_) {
983 connectionEventCallback_->onBackoffEnded();
988 // Register the handler.
989 for (auto& handler : sockets_) {
990 if (!handler.registerHandler(
991 EventHandler::READ | EventHandler::PERSIST)) {
992 // We're hosed. We could just re-schedule backoffTimeout_ to
993 // re-try again after a little bit. However, we don't want to
994 // loop retrying forever if we can't re-enable accepts. Just
995 // abort the entire program in this state; things are really bad
996 // and restarting the entire server is probably the best remedy.
998 << "failed to re-enable AsyncServerSocket accepts after backoff; "
1003 if (connectionEventCallback_) {
1004 connectionEventCallback_->onBackoffEnded();