proxygen: folly::AsyncServerSocket Class Reference

#include <AsyncServerSocket.h>

Inheritance diagram for folly::AsyncServerSocket:

Classes
class	AcceptCallback

class	BackoffTimeout

struct	CallbackInfo

class	ConnectionEventCallback

struct	QueueMessage

class	RemoteAcceptor

struct	ServerEventHandler

Public Types
typedef std::unique_ptr< AsyncServerSocket, Destructor >	UniquePtr

Public Member Functions
	AsyncServerSocket (AsyncServerSocket &&)=delete

	AsyncServerSocket (EventBase *eventBase=nullptr)

void	setShutdownSocketSet (const std::weak_ptr< ShutdownSocketSet > &wNewSS)

void	destroy () override

void	attachEventBase (EventBase *eventBase)

void	detachEventBase ()

EventBase *	getEventBase () const override

void	useExistingSocket (int fd)

void	useExistingSockets (const std::vector< int > &fds)

std::vector< int >	getSockets () const

int	getSocket () const

bool	setZeroCopy (bool enable)

virtual void	bind (const SocketAddress &address)

virtual void	bind (const std::vector< IPAddress > &ipAddresses, uint16_t port)

virtual void	bind (uint16_t port)

void	getAddress (SocketAddress *addressReturn) const override

SocketAddress	getAddress () const

std::vector< SocketAddress >	getAddresses () const

virtual void	listen (int backlog)

virtual void	addAcceptCallback (AcceptCallback callback, EventBase eventBase, uint32_t maxAtOnce=kDefaultCallbackAcceptAtOnce)

void	removeAcceptCallback (AcceptCallback callback, EventBase eventBase)

virtual void	startAccepting ()

void	pauseAccepting ()

int	stopAccepting (int shutdownFlags=-1)

uint32_t	getMaxAcceptAtOnce () const

void	setMaxAcceptAtOnce (uint32_t numConns)

uint32_t	getMaxNumMessagesInQueue () const

void	setMaxNumMessagesInQueue (uint32_t num)

double	getAcceptRateAdjustSpeed () const

void	setAcceptRateAdjustSpeed (double speed)

void	setTosReflect (bool enable)

bool	getTosReflect ()

std::size_t	getNumDroppedConnections () const

int64_t	getNumPendingMessagesInQueue () const

void	setKeepAliveEnabled (bool enabled)

bool	getKeepAliveEnabled () const

void	setReusePortEnabled (bool enabled)

bool	getReusePortEnabled_ () const

void	setCloseOnExec (bool closeOnExec)

bool	getCloseOnExec () const

void	setTFOEnabled (bool enabled, uint32_t maxTFOQueueSize)

void	disableTransparentTls ()

bool	getAccepting () const

void	setConnectionEventCallback (ConnectionEventCallback *const connectionEventCallback)

ConnectionEventCallback *	getConnectionEventCallback () const

Public Member Functions inherited from folly::DelayedDestruction
bool	getDestroyPending () const

Public Member Functions inherited from folly::DelayedDestructionBase
virtual	~DelayedDestructionBase ()=default

Public Member Functions inherited from folly::AsyncSocketBase
virtual	~AsyncSocketBase ()=default

Static Public Member Functions
static std::shared_ptr< AsyncServerSocket >	newSocket (EventBase *evb=nullptr)

Static Public Attributes
static const uint32_t	kDefaultMaxAcceptAtOnce = 30

static const uint32_t	kDefaultCallbackAcceptAtOnce = 5

static const uint32_t	kDefaultMaxMessagesInQueue = 1024

Protected Member Functions
	~AsyncServerSocket () override

Protected Member Functions inherited from folly::DelayedDestruction
	~DelayedDestruction () override=default

	DelayedDestruction ()

Protected Member Functions inherited from folly::DelayedDestructionBase
	DelayedDestructionBase ()

uint32_t	getDestructorGuardCount () const

Private Types
enum	MessageType { MessageType::MSG_NEW_CONN = 0, MessageType::MSG_ERROR = 1 }

Private Member Functions
virtual void	handlerReady (uint16_t events, int socket, sa_family_t family) noexcept

int	createSocket (int family)

void	setupSocket (int fd, int family)

void	bindSocket (int fd, const SocketAddress &address, bool isExistingSocket)

void	dispatchSocket (int socket, SocketAddress &&address)

void	dispatchError (const char *msg, int errnoValue)

void	enterBackoff ()

void	backoffTimeoutExpired ()

CallbackInfo *	nextCallback ()

Private Attributes
EventBase *	eventBase_

std::vector< ServerEventHandler >	sockets_

std::vector< int >	pendingCloseSockets_

bool	accepting_

uint32_t	maxAcceptAtOnce_

uint32_t	maxNumMsgsInQueue_

double	acceptRateAdjustSpeed_

double	acceptRate_

std::chrono::time_point< std::chrono::steady_clock >	lastAccepTimestamp_

std::size_t	numDroppedConnections_

uint32_t	callbackIndex_

BackoffTimeout *	backoffTimeout_

std::vector< CallbackInfo >	callbacks_

bool	keepAliveEnabled_

bool	reusePortEnabled_ {false}

bool	closeOnExec_

bool	tfo_ {false}

bool	noTransparentTls_ {false}

uint32_t	tfoMaxQueueSize_ {0}

std::weak_ptr< ShutdownSocketSet >	wShutdownSocketSet_

ConnectionEventCallback *	connectionEventCallback_ {nullptr}

bool	tosReflect_ {false}

Detailed Description

A listening socket that asynchronously informs a callback whenever a new connection has been accepted.

Unlike most async interfaces that always invoke their callback in the same EventBase thread, AsyncServerSocket is unusual in that it can distribute the callbacks across multiple EventBase threads.

This supports a common use case for network servers to distribute incoming connections across a number of EventBase threads. (Servers typically run with one EventBase thread per CPU.)

Despite being able to invoke callbacks in multiple EventBase threads, AsyncServerSocket still has one "primary" EventBase. Operations that modify the AsyncServerSocket state may only be performed from the primary EventBase thread.

Definition at line 65 of file AsyncServerSocket.h.

Member Typedef Documentation

typedef std::unique_ptr<AsyncServerSocket, Destructor> folly::AsyncServerSocket::UniquePtr

Definition at line 67 of file AsyncServerSocket.h.

Member Enumeration Documentation

enum folly::AsyncServerSocket::MessageType

strongprivate

Enumerator
MSG_NEW_CONN
MSG_ERROR

Definition at line 742 of file AsyncServerSocket.h.

742 { MSG_NEW_CONN = 0, MSG_ERROR = 1 };

Constructor & Destructor Documentation

folly::AsyncServerSocket::AsyncServerSocket ( AsyncServerSocket && )

delete

Referenced by newSocket().

folly::AsyncServerSocket::AsyncServerSocket ( EventBase * eventBase = nullptr )

explicit

Create a new AsyncServerSocket with the specified EventBase.

Parameters

eventBase The EventBase to use for driving the asynchronous I/O. If this parameter is nullptr, attachEventBase() must be called before this socket can begin accepting connections.

Definition at line 162 of file AsyncServerSocket.cpp.

References disableTransparentTls().

     : 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) {
   disableTransparentTls();
 }

folly::AsyncServerSocket::~AsyncServerSocket ( )

overrideprotected

Protected destructor.

Invoke destroy() instead to destroy the AsyncServerSocket.

Definition at line 203 of file AsyncServerSocket.cpp.

References callbacks_.

Referenced by getConnectionEventCallback().

                                       {
   assert(callbacks_.empty());
 }

Member Function Documentation

void folly::AsyncServerSocket::addAcceptCallback	(	AcceptCallback *	callback,
		EventBase *	eventBase,
		uint32_t	maxAtOnce = `kDefaultCallbackAcceptAtOnce`
	)

virtual

Add an AcceptCallback.

When a new socket is accepted, one of the AcceptCallbacks will be invoked with the new socket. The AcceptCallbacks are invoked in a round-robin fashion. This allows the accepted sockets to be distributed among a pool of threads, each running its own EventBase object. This is a common model, since most asynchronous-style servers typically run one EventBase thread per CPU.

The EventBase object associated with each AcceptCallback must be running its loop. If the EventBase loop is not running, sockets will still be scheduled for the callback, but the callback cannot actually get invoked until the loop runs.

This method must be invoked from the AsyncServerSocket's primary EventBase thread.

Note that startAccepting() must be called on the AsyncServerSocket to cause it to actually start accepting sockets once callbacks have been installed.

Parameters

callback	The callback to invoke.
eventBase	The EventBase to use to invoke the callback. This parameter may be nullptr, in which case the callback will be invoked in the AsyncServerSocket's primary EventBase.
maxAtOnce	The maximum number of connections to accept in this callback on a single iteration of the event base loop. This only takes effect when eventBase is non-nullptr. When using a nullptr eventBase for the callback, the setMaxAcceptAtOnce() method controls how many connections the main event base will accept at once.

Definition at line 586 of file AsyncServerSocket.cpp.

References accepting_, folly::AsyncServerSocket::AcceptCallback::acceptStarted(), callbacks_, connectionEventCallback_, folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, maxNumMsgsInQueue_, folly::AsyncServerSocket::RemoteAcceptor::start(), and startAccepting().

Referenced by folly::ZeroCopyTestServer::addCallbackToServerSocket(), getAddress(), wangle::Acceptor::init(), and serverSocketSanityTest().

                         {
   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) {
     // Run in AsyncServerSocket's eventbase; notify that we are
     // starting to accept connections
     callback->acceptStarted();
     return;
   }
 
   // Start the remote acceptor.
   //
   // It would be nice if we could avoid starting the remote acceptor if
   // eventBase == eventBase_.  However, that would cause issues if
   // detachEventBase() and attachEventBase() were ever used to change the
   // primary EventBase for the server socket.  Therefore we require the caller
   // to specify a nullptr EventBase if they want to ensure that the callback is
   // always invoked in the primary EventBase, and to be able to invoke that
   // callback more efficiently without having to use a notification queue.
   RemoteAcceptor* acceptor = nullptr;
   try {
     acceptor = new RemoteAcceptor(callback, connectionEventCallback_);
     acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
   } catch (...) {
     callbacks_.pop_back();
     delete acceptor;
     throw;
   }
   callbacks_.back().consumer = acceptor;
 }

void folly::AsyncServerSocket::attachEventBase ( EventBase * eventBase )

Attach this AsyncServerSocket to its primary EventBase.

This may only be called if the AsyncServerSocket is not already attached to a EventBase. The AsyncServerSocket must be attached to a EventBase before it can begin accepting connections.

Definition at line 274 of file AsyncServerSocket.cpp.

References folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), and sockets_.

Referenced by newSocket(), and folly::AsyncServerSocket::ServerEventHandler::operator=().

                                                             {
   assert(eventBase_ == nullptr);
   eventBase->dcheckIsInEventBaseThread();
 
   eventBase_ = eventBase;
   for (auto& handler : sockets_) {
     handler.attachEventBase(eventBase);
   }
 }

void folly::AsyncServerSocket::backoffTimeoutExpired ( )

private

Definition at line 1142 of file AsyncServerSocket.cpp.

References accepting_, callbacks_, connectionEventCallback_, folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), folly::AsyncServerSocket::ConnectionEventCallback::onBackoffEnded(), folly::EventHandler::PERSIST, folly::EventHandler::READ, and sockets_.

                                               {
   // accepting_ should still be true.
   // If pauseAccepting() was called while in the backoff state it will cancel
   // the backoff timeout.
   assert(accepting_);
   // We can't be detached from the EventBase without being paused
   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;
   }
 
   // Register the handler.
   for (auto& handler : sockets_) {
     if (!handler.registerHandler(EventHandler::READ | EventHandler::PERSIST)) {
       // We're hosed.  We could just re-schedule backoffTimeout_ to
       // re-try again after a little bit.  However, we don't want to
       // loop retrying forever if we can't re-enable accepts.  Just
       // abort the entire program in this state; things are really bad
       // and restarting the entire server is probably the best remedy.
       LOG(ERROR)
           << "failed to re-enable AsyncServerSocket accepts after backoff; "
           << "crashing now";
       abort();
     }
   }
   if (connectionEventCallback_) {
     connectionEventCallback_->onBackoffEnded();
   }
 }

void folly::AsyncServerSocket::bind ( const SocketAddress & address )

virtual

Bind to the specified address.

This must be called from the primary EventBase thread.

Throws TTransportException on error.

Definition at line 372 of file AsyncServerSocket.cpp.

References bindSocket(), createSocket(), folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, folly::SocketAddress::getFamily(), and sockets_.

Referenced by getSocket().

                                                          {
   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.
   // Don't set socket_ yet, so that socket_ will remain uninitialized if an
   // error occurs.
   int fd;
   if (sockets_.size() == 0) {
     fd = createSocket(address.getFamily());
   } else if (sockets_.size() == 1) {
     if (address.getFamily() != sockets_[0].addressFamily_) {
       throw std::invalid_argument(
           "Attempted to bind address to socket with "
           "different address family");
     }
     fd = sockets_[0].socket_;
   } else {
     throw std::invalid_argument("Attempted to bind to multiple fds");
   }
 
   bindSocket(fd, address, !sockets_.empty());
 }

void folly::AsyncServerSocket::bind	(	const std::vector< IPAddress > &	ipAddresses,
		uint16_t	port
	)

virtual

Bind to the specified port for the specified addresses.

This must be called from the primary EventBase thread.

Throws TTransportException on error.

Definition at line 398 of file AsyncServerSocket.cpp.

References bindSocket(), createSocket(), and sockets_.

                    {
   if (ipAddresses.empty()) {
     throw std::invalid_argument("No ip addresses were provided");
   }
   if (!sockets_.empty()) {
     throw std::invalid_argument(
         "Cannot call bind on a AsyncServerSocket "
         "that already has a socket.");
   }
 
   for (const IPAddress& ipAddress : ipAddresses) {
     SocketAddress address(ipAddress.toFullyQualified(), port);
     int fd = createSocket(address.getFamily());
 
     bindSocket(fd, address, false);
   }
   if (sockets_.size() == 0) {
     throw std::runtime_error(
         "did not bind any async server socket for port and addresses");
   }
 }

void folly::AsyncServerSocket::bind ( uint16_t port )

virtual

Bind to the specified port.

This must be called from the primary EventBase thread.

Throws TTransportException on error.

Definition at line 422 of file AsyncServerSocket.cpp.

References folly::netops::bind(), folly::ShutdownSocketSet::close(), folly::closeNoInt(), eventBase_, folly::SocketAddress::getFamily(), folly::SocketAddress::getFamilyNameFrom(), folly::SocketAddress::getPort(), folly::SocketAddress::getPortFrom(), folly::kIsWindows, noTransparentTls_, s, SCOPE_EXIT, folly::SocketAddress::setFromLocalAddress(), folly::netops::setsockopt(), setupSocket(), folly::test::shutdownSocketSet, folly::netops::socket(), sockets_, folly::throwSystemError(), and wShutdownSocketSet_.

                                           {
   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 | AI_NUMERICSERV;
   snprintf(sport, sizeof(sport), "%u", port);
 
   // 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");
   }
 
   SCOPE_EXIT {
     freeaddrinfo(res0);
   };
 
   auto setupAddress = [&](struct addrinfo* res) {
     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;
     }
     CHECK_GE(s, 0);
 
     try {
       setupSocket(s, res->ai_family);
     } catch (...) {
       closeNoInt(s);
       throw;
     }
 
     if (res->ai_family == AF_INET6) {
       int v6only = 1;
       CHECK(
           0 ==
           setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(v6only)));
     }
 
     // Bind to the socket
     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 ",
           SocketAddress::getPortFrom(res->ai_addr),
           " family ",
           SocketAddress::getFamilyNameFrom(res->ai_addr, "<unknown>"));
     }
 
 #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 = 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
     // getaddrinfo(nullptr) with AI_PASSIVE returns:
     // - 0.0.0.0 (IPv4-only)
     // - :: (IPv6+IPv4) in this order
     // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
     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.
     if (sockets_.size() == 1 && port == 0) {
       SocketAddress address;
       address.setFromLocalAddress(sockets_.back().socket_);
       snprintf(sport, sizeof(sport), "%u", address.getPort());
       freeaddrinfo(res0);
       CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
     }
 
     try {
       for (struct addrinfo* res = res0; res; res = res->ai_next) {
         if (res->ai_family != AF_INET6) {
           setupAddress(res);
         }
       }
     } 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 (
               const auto shutdownSocketSet = wShutdownSocketSet_.lock()) {
             shutdownSocketSet->close(socket.socket_);
           } else {
             closeNoInt(socket.socket_);
           }
         }
         sockets_.clear();
         snprintf(sport, sizeof(sport), "%u", port);
         freeaddrinfo(res0);
         CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
         continue;
       }
 
       throw;
     }
 
     break;
   }
 
   if (sockets_.size() == 0) {
     throw std::runtime_error("did not bind any async server socket for port");
   }
 }

void folly::AsyncServerSocket::bindSocket	(	int	fd,
		const SocketAddress &	address,
		bool	isExistingSocket
	)

private

Definition at line 331 of file AsyncServerSocket.cpp.

References folly::netops::bind(), folly::closeNoInt(), folly::SocketAddress::describe(), eventBase_, folly::SocketAddress::getActualSize(), folly::SocketAddress::getAddress(), folly::SocketAddress::getFamily(), noTransparentTls_, folly::netops::setsockopt(), sockets_, and folly::throwSystemError().

Referenced by bind().

                            {
   sockaddr_storage addrStorage;
   address.getAddress(&addrStorage);
   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 
   if (fsp::bind(fd, saddr, address.getActualSize()) != 0) {
     if (!isExistingSocket) {
       closeNoInt(fd);
     }
     folly::throwSystemError(
         errno, "failed to bind to async server socket: " + 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_.emplace_back(eventBase_, fd, this, address.getFamily());
   }
 }

int folly::AsyncServerSocket::createSocket ( int family )

private

Definition at line 738 of file AsyncServerSocket.cpp.

References folly::closeNoInt(), setupSocket(), folly::netops::socket(), and folly::throwSystemError().

Referenced by bind().

                                               {
   int fd = fsp::socket(family, SOCK_STREAM, 0);
   if (fd == -1) {
     folly::throwSystemError(errno, "error creating async server socket");
   }
 
   try {
     setupSocket(fd, family);
   } catch (...) {
     closeNoInt(fd);
     throw;
   }
   return fd;
 }

void folly::AsyncServerSocket::destroy ( )

overridevirtual

Destroy the socket.

AsyncServerSocket::destroy() must be called to destroy the socket. The normal destructor is private, and should not be invoked directly. This prevents callers from deleting a AsyncServerSocket while it is invoking a callback.

destroy() must be invoked from the socket's primary EventBase thread.

If there are AcceptCallbacks still installed when destroy() is called, acceptStopped() will be called on these callbacks to notify them that accepting has stopped. Accept callbacks being driven by other EventBase threads may continue to receive new accept callbacks for a brief period of time after destroy() returns. They will not receive any more callback invocations once acceptStopped() is invoked.

Reimplemented from folly::DelayedDestruction.

Definition at line 264 of file AsyncServerSocket.cpp.

References folly::closeNoInt(), folly::DelayedDestruction::destroy(), pendingCloseSockets_, s, and stopAccepting().

Referenced by newSocket(), and wangle::AsyncServerSocketFactory::ThreadSafeDestructor::operator()().

                                 {
   stopAccepting();
   for (auto s : pendingCloseSockets_) {
     closeNoInt(s);
   }
   // Then call DelayedDestruction::destroy() to take care of
   // whether or not we need immediate or delayed destruction
   DelayedDestruction::destroy();
 }

void folly::AsyncServerSocket::detachEventBase ( )

Detach the AsyncServerSocket from its primary EventBase.

detachEventBase() may only be called if the AsyncServerSocket is not currently accepting connections.

Definition at line 284 of file AsyncServerSocket.cpp.

References accepting_, folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), and sockets_.

Referenced by newSocket(), and folly::AsyncServerSocket::ServerEventHandler::operator=().

                                         {
   assert(eventBase_ != nullptr);
   eventBase_->dcheckIsInEventBaseThread();
   assert(!accepting_);
 
   eventBase_ = nullptr;
   for (auto& handler : sockets_) {
     handler.detachEventBase();
   }
 }

void folly::AsyncServerSocket::disableTransparentTls ( )

inline

Do not attempt the transparent TLS handshake

Definition at line 707 of file AsyncServerSocket.h.

References noTransparentTls_.

Referenced by AsyncServerSocket().

                                {
     noTransparentTls_ = true;
   }

void folly::AsyncServerSocket::dispatchError	(	const char *	msg,
		int	errnoValue
	)

private

Definition at line 1056 of file AsyncServerSocket.cpp.

References folly::AsyncServerSocket::AcceptCallback::acceptError(), folly::AsyncServerSocket::CallbackInfo::callback, callbackIndex_, folly::AsyncServerSocket::CallbackInfo::consumer, folly::AsyncServerSocket::QueueMessage::err, errnoValue, folly::AsyncServerSocket::CallbackInfo::eventBase, folly::AsyncServerSocket::RemoteAcceptor::getQueue(), deadlock::info(), folly::gen::move, folly::AsyncServerSocket::QueueMessage::msg, MSG_ERROR, nextCallback(), string, folly::AsyncServerSocket::QueueMessage::type, and uint32_t.

Referenced by handlerReady().

                                                                         {
   uint32_t startingIndex = callbackIndex_;
   CallbackInfo* info = nextCallback();
 
   // Create a message to send over the notification queue
   QueueMessage msg;
   msg.type = MessageType::MSG_ERROR;
   msg.err = errnoValue;
   msg.msg = std::move(msgstr);
 
   while (true) {
     // Short circuit if the callback is in the primary EventBase thread
     if (info->eventBase == nullptr || info->eventBase == this->eventBase_) {
       std::runtime_error ex(
           std::string(msgstr) + folly::to<std::string>(errnoValue));
       info->callback->acceptError(ex);
       return;
     }
 
     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
       return;
     }
     // Fall through and try another callback
 
     if (callbackIndex_ == startingIndex) {
       // The notification queues for all of the callbacks were full.
       // We can't really do anything at this point.
       LOG_EVERY_N(ERROR, 100)
           << "failed to dispatch accept error: all accept"
           << " callback queues are full: error msg:  " << msg.msg << ": "
           << errnoValue;
       return;
     }
     info = nextCallback();
   }
 }

void folly::AsyncServerSocket::dispatchSocket	(	int	socket,
		SocketAddress &&	address
	)

private

Definition at line 995 of file AsyncServerSocket.cpp.

Referenced by handlerReady().

                                                                           {
   uint32_t startingIndex = callbackIndex_;
 
   // Short circuit if the callback is in the primary EventBase thread
 
   CallbackInfo* info = nextCallback();
   if (info->eventBase == nullptr || info->eventBase == this->eventBase_) {
     info->callback->connectionAccepted(socket, address);
     return;
   }
 
   const SocketAddress addr(address);
   // Create a message to send over the notification queue
   QueueMessage msg;
   msg.type = MessageType::MSG_NEW_CONN;
   msg.address = std::move(address);
   msg.fd = 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
 
     ++numDroppedConnections_;
     if (acceptRateAdjustSpeed_ > 0) {
       // aggressively decrease accept rate when in trouble
       static const double kAcceptRateDecreaseSpeed = 0.1;
       acceptRate_ *= 1 - kAcceptRateDecreaseSpeed;
     }
 
     if (callbackIndex_ == startingIndex) {
       // The notification queue was full
       // We can't really do anything at this point other than close the socket.
       //
       // This should only happen if a user's service is behaving extremely
       // badly and none of the EventBase threads are looping fast enough to
       // process the incoming connections.  If the service is overloaded, it
       // 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_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;
     }
 
     info = nextCallback();
   }
 }

void folly::AsyncServerSocket::enterBackoff ( )

private

Definition at line 1093 of file AsyncServerSocket.cpp.

References backoffTimeout_, connectionEventCallback_, handler(), folly::AsyncServerSocket::ConnectionEventCallback::onBackoffError(), folly::AsyncServerSocket::ConnectionEventCallback::onBackoffStarted(), folly::AsyncTimeout::scheduleTimeout(), sockets_, and uint32_t.

Referenced by handlerReady().

                                      {
   // If this is the first time we have entered the backoff state,
   // allocate backoffTimeout_.
   if (backoffTimeout_ == nullptr) {
     try {
       backoffTimeout_ = new BackoffTimeout(this);
     } 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;
     }
   }
 
   // For now, we simply pause accepting for 1 second.
   //
   // We could add some smarter backoff calculation here in the future.  (e.g.,
   // start sleeping for longer if we keep hitting the backoff frequently.)
   // Typically the user needs to figure out why the server is overloaded and
   // fix it in some other way, though.  The backoff timer is just a simple
   // mechanism to try and give the connection processing code a little bit of
   // breathing room to catch up, and to avoid just spinning and failing to
   // accept over and over again.
   const uint32_t timeoutMS = 1000;
   if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
     LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
                << "unable to temporarly pause accepting";
     if (connectionEventCallback_) {
       connectionEventCallback_->onBackoffError();
     }
     return;
   }
 
   // The backoff timer is scheduled to re-enable accepts.
   // Go ahead and disable accepts for now.  We leave accepting_ set to true,
   // since that tracks the desired state requested by the user.
   for (auto& handler : sockets_) {
     handler.unregisterHandler();
   }
   if (connectionEventCallback_) {
     connectionEventCallback_->onBackoffStarted();
   }
 }

bool folly::AsyncServerSocket::getAccepting ( ) const

inline

Get whether or not the socket is accepting new connections

Definition at line 714 of file AsyncServerSocket.h.

References accepting_.

                             {
     return accepting_;
   }

double folly::AsyncServerSocket::getAcceptRateAdjustSpeed ( ) const

inline

Get the speed of adjusting connection accept rate.

Definition at line 569 of file AsyncServerSocket.h.

References acceptRateAdjustSpeed_.

                                           {
     return acceptRateAdjustSpeed_;
   }

void folly::AsyncServerSocket::getAddress ( SocketAddress * addressReturn ) const

overridevirtual

Get the local address to which the socket is bound.

Throws TTransportException on error.

Implements folly::AsyncSocketBase.

Definition at line 567 of file AsyncServerSocket.cpp.

References folly::SocketAddress::setFromLocalAddress(), socket_, and sockets_.

Referenced by serverSocketSanityTest().

                                                                      {
   CHECK(sockets_.size() >= 1);
   VLOG_IF(2, sockets_.size() > 1)
       << "Warning: getAddress() called and multiple addresses available ("
       << sockets_.size() << "). Returning only the first one.";
 
   addressReturn->setFromLocalAddress(sockets_[0].socket_);
 }

SocketAddress folly::AsyncServerSocket::getAddress ( ) const

inline

Get the local address to which the socket is bound.

Throws TTransportException on error.

Definition at line 368 of file AsyncServerSocket.h.

References addAcceptCallback(), getAddresses(), listen(), pauseAccepting(), removeAcceptCallback(), startAccepting(), stopAccepting(), and uint32_t.

Referenced by getSocket().

                                    {
     SocketAddress ret;
     getAddress(&ret);
     return ret;
   }

std::vector< SocketAddress > folly::AsyncServerSocket::getAddresses ( ) const

Get all the local addresses to which the socket is bound.

Throws TTransportException on error.

Definition at line 576 of file AsyncServerSocket.cpp.

References folly::netops::socket(), and sockets_.

Referenced by getAddress().

                                                                {
   CHECK(sockets_.size() >= 1);
   auto tsaVec = std::vector<SocketAddress>(sockets_.size());
   auto tsaIter = tsaVec.begin();
   for (const auto& socket : sockets_) {
     (tsaIter++)->setFromLocalAddress(socket.socket_);
   };
   return tsaVec;
 }

bool folly::AsyncServerSocket::getCloseOnExec ( ) const

inline

Get whether or not FD_CLOEXEC is enabled on the server socket.

Definition at line 692 of file AsyncServerSocket.h.

References closeOnExec_.

                               {
     return closeOnExec_;
   }

ConnectionEventCallback* folly::AsyncServerSocket::getConnectionEventCallback ( ) const

inline

Get the ConnectionEventCallback

Definition at line 729 of file AsyncServerSocket.h.

References connectionEventCallback_, and ~AsyncServerSocket().

                                                               {
     return connectionEventCallback_;
   }

EventBase* folly::AsyncServerSocket::getEventBase ( ) const

inlineoverridevirtual

Get the EventBase used by this socket.

Implements folly::AsyncSocketBase.

Definition at line 273 of file AsyncServerSocket.h.

References eventBase_, useExistingSocket(), and useExistingSockets().

Referenced by wangle::AsyncServerSocketFactory::ThreadSafeDestructor::operator()(), and serverSocketSanityTest().

                                            {
     return eventBase_;
   }

bool folly::AsyncServerSocket::getKeepAliveEnabled ( ) const

inline

Get whether or not SO_KEEPALIVE is enabled on the server socket.

Definition at line 647 of file AsyncServerSocket.h.

References keepAliveEnabled_.

                                    {
     return keepAliveEnabled_;
   }

uint32_t folly::AsyncServerSocket::getMaxAcceptAtOnce ( ) const

inline

Get the maximum number of connections that will be accepted each time around the event loop.

Definition at line 525 of file AsyncServerSocket.h.

References maxAcceptAtOnce_.

                                       {
     return maxAcceptAtOnce_;
   }

uint32_t folly::AsyncServerSocket::getMaxNumMessagesInQueue ( ) const

inline

Get the maximum number of unprocessed messages which a NotificationQueue can hold.

Definition at line 551 of file AsyncServerSocket.h.

References maxNumMsgsInQueue_.

                                             {
     return maxNumMsgsInQueue_;
   }

std::size_t folly::AsyncServerSocket::getNumDroppedConnections ( ) const

inline

Get the number of connections dropped by the AsyncServerSocket

Definition at line 592 of file AsyncServerSocket.h.

References numDroppedConnections_.

                                              {
     return numDroppedConnections_;
   }

int64_t folly::AsyncServerSocket::getNumPendingMessagesInQueue ( ) const

inline

Get the current number of unprocessed messages in NotificationQueue.

This method must be invoked from the AsyncServerSocket's primary EventBase thread. Use EventBase::runInEventBaseThread() to schedule the operation in the correct EventBase if your code is not in the server socket's primary EventBase.

Definition at line 604 of file AsyncServerSocket.h.

References callbacks_, folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, and int64_t.

                                                {
     if (eventBase_) {
       eventBase_->dcheckIsInEventBaseThread();
     }
     int64_t numMsgs = 0;
     for (const auto& callback : callbacks_) {
       if (callback.consumer) {
         numMsgs += callback.consumer->getQueue()->size();
       }
     }
     return numMsgs;
   }

bool folly::AsyncServerSocket::getReusePortEnabled_ ( ) const

inline

Get whether or not SO_REUSEPORT is enabled on the server socket.

Definition at line 677 of file AsyncServerSocket.h.

References reusePortEnabled_.

                                     {
     return reusePortEnabled_;
   }

int folly::AsyncServerSocket::getSocket ( ) const

inline

Backwards compatible getSocket, warns if > 1 socket

Definition at line 312 of file AsyncServerSocket.h.

References bind(), getAddress(), setZeroCopy(), sockets_, and uint16_t.

Referenced by setZeroCopy().

                         {
     if (sockets_.size() > 1) {
       VLOG(2) << "Warning: getSocket can return multiple fds, "
               << "but getSockets was not called, so only returning the first";
     }
     if (sockets_.size() == 0) {
       return -1;
     } else {
       return sockets_[0].socket_;
     }
   }

std::vector<int> folly::AsyncServerSocket::getSockets ( ) const

inline

Return the underlying file descriptor

Definition at line 301 of file AsyncServerSocket.h.

References handler(), and sockets_.

Referenced by proxygen::HTTPServer::getListenSocket(), and wangle::Acceptor::init().

                                     {
     std::vector<int> sockets;
     for (auto& handler : sockets_) {
       sockets.push_back(handler.socket_);
     }
     return sockets;
   }

bool folly::AsyncServerSocket::getTosReflect ( )

inline

Definition at line 585 of file AsyncServerSocket.h.

References tosReflect_.

                        {
     return tosReflect_;
   }

void folly::AsyncServerSocket::handlerReady	(	uint16_t	events,
		int	socket,
		sa_family_t	family
	)

privatevirtualnoexcept

Definition at line 855 of file AsyncServerSocket.cpp.

References folly::netops::accept(), accepting_, acceptRate_, acceptRateAdjustSpeed_, folly::Endian::big(), buffer(), callbacks_, folly::closeNoInt(), connectionEventCallback_, dispatchError(), dispatchSocket(), enterBackoff(), folly::netops::getsockopt(), folly::kIsLinux, lastAccepTimestamp_, maxAcceptAtOnce_, folly::gen::move, now(), numDroppedConnections_, folly::AsyncServerSocket::ConnectionEventCallback::onConnectionAccepted(), folly::AsyncServerSocket::ConnectionEventCallback::onConnectionAcceptError(), folly::AsyncServerSocket::ConnectionEventCallback::onConnectionDropped(), folly::SocketAddress::setFromSockaddr(), folly::netops::setsockopt(), TCP_SAVED_SYN, tosReflect_, and uint32_t.

                                         {
   assert(!callbacks_.empty());
   DestructorGuard dg(this);
 
   // Only accept up to maxAcceptAtOnce_ connections at a time,
   // to avoid starving other I/O handlers using this EventBase.
   for (uint32_t n = 0; n < maxAcceptAtOnce_; ++n) {
     SocketAddress address;
 
     sockaddr_storage addrStorage;
     socklen_t addrLen = sizeof(addrStorage);
     sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 
     // In some cases, accept() doesn't seem to update these correctly.
     saddr->sa_family = addressFamily;
     if (addressFamily == AF_UNIX) {
       addrLen = sizeof(struct sockaddr_un);
     }
 
     // Accept a new client socket
 #ifdef SOCK_NONBLOCK
     int clientSocket = accept4(fd, saddr, &addrLen, SOCK_NONBLOCK);
 #else
     int clientSocket = accept(fd, saddr, &addrLen);
 #endif
 
     address.setFromSockaddr(saddr, addrLen);
 
     if (clientSocket >= 0 && connectionEventCallback_) {
       connectionEventCallback_->onConnectionAccepted(clientSocket, address);
     }
 
     // Connection accepted, get the SYN packet from the client if
     // TOS reflect is enabled
     if (kIsLinux && clientSocket >= 0 && tosReflect_) {
       std::array<uint32_t, 64> buffer;
       socklen_t len = sizeof(buffer);
       int ret =
           getsockopt(clientSocket, IPPROTO_TCP, TCP_SAVED_SYN, &buffer, &len);
 
       if (ret == 0) {
         uint32_t tosWord = folly::Endian::big(buffer[0]);
         if (addressFamily == AF_INET6) {
           tosWord = (tosWord & 0x0FC00000) >> 20;
           ret = setsockopt(
               clientSocket,
               IPPROTO_IPV6,
               IPV6_TCLASS,
               &tosWord,
               sizeof(tosWord));
         } else if (addressFamily == AF_INET) {
           tosWord = (tosWord & 0x00FC0000) >> 16;
           ret = setsockopt(
               clientSocket, IPPROTO_IP, IP_TOS, &tosWord, sizeof(tosWord));
         }
 
         if (ret != 0) {
           LOG(ERROR) << "Unable to set TOS for accepted socket "
                      << clientSocket;
         }
       } else {
         LOG(ERROR) << "Unable to get SYN packet for accepted socket "
                    << clientSocket;
       }
     }
 
     std::chrono::time_point<std::chrono::steady_clock> nowMs =
         std::chrono::steady_clock::now();
     auto timeSinceLastAccept = std::max<int64_t>(
         0,
         nowMs.time_since_epoch().count() -
             lastAccepTimestamp_.time_since_epoch().count());
     lastAccepTimestamp_ = nowMs;
     if (acceptRate_ < 1) {
       acceptRate_ *= 1 + acceptRateAdjustSpeed_ * timeSinceLastAccept;
       if (acceptRate_ >= 1) {
         acceptRate_ = 1;
       } else if (rand() > acceptRate_ * RAND_MAX) {
         ++numDroppedConnections_;
         if (clientSocket >= 0) {
           closeNoInt(clientSocket);
           if (connectionEventCallback_) {
             connectionEventCallback_->onConnectionDropped(
                 clientSocket, address);
           }
         }
         continue;
       }
     }
 
     if (clientSocket < 0) {
       if (errno == EAGAIN) {
         // No more sockets to accept right now.
         // Check for this code first, since it's the most common.
         return;
       } else if (errno == EMFILE || errno == ENFILE) {
         // We're out of file descriptors.  Perhaps we're accepting connections
         // too quickly. Pause accepting briefly to back off and give the server
         // a chance to recover.
         LOG(ERROR) << "accept failed: out of file descriptors; entering accept "
                       "back-off state";
         enterBackoff();
 
         // Dispatch the error message
         dispatchError("accept() failed", errno);
       } else {
         dispatchError("accept() failed", errno);
       }
       if (connectionEventCallback_) {
         connectionEventCallback_->onConnectionAcceptError(errno);
       }
       return;
     }
 
 #ifndef SOCK_NONBLOCK
     // Explicitly set the new connection to non-blocking mode
     if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
       closeNoInt(clientSocket);
       dispatchError(
           "failed to set accepted socket to non-blocking mode", errno);
       if (connectionEventCallback_) {
         connectionEventCallback_->onConnectionDropped(clientSocket, address);
       }
       return;
     }
 #endif
 
     // Inform the callback about the new connection
     dispatchSocket(clientSocket, std::move(address));
 
     // If we aren't accepting any more, break out of the loop
     if (!accepting_ || callbacks_.empty()) {
       break;
     }
   }
 }

void folly::AsyncServerSocket::listen ( int backlog )

virtual

Begin listening for connections.

This calls listen() with the specified backlog.

Once listen() is invoked the socket will actually be open so that remote clients may establish connections. (Clients that attempt to connect before listen() is called will receive a connection refused error.)

At least one callback must be set and startAccepting() must be called to actually begin notifying the accept callbacks of newly accepted connections. The backlog parameter controls how many connections the kernel will accept and buffer internally while the accept callbacks are paused (or if accepting is enabled but the callbacks cannot keep up).

bind() must be called before calling listen(). listen() must be called from the primary EventBase thread.

Throws TTransportException on error.

Definition at line 554 of file AsyncServerSocket.cpp.

References folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), folly::netops::listen(), sockets_, and folly::throwSystemError().

Referenced by getAddress().

                                           {
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
 
   // Start listening
   for (auto& handler : sockets_) {
     if (fsp::listen(handler.socket_, backlog) == -1) {
       folly::throwSystemError(errno, "failed to listen on async server socket");
     }
   }
 }

static std::shared_ptr<AsyncServerSocket> folly::AsyncServerSocket::newSocket ( EventBase * evb = nullptr )

inlinestatic

Helper function to create a shared_ptr<AsyncServerSocket>.

This passes in the correct destructor object, since AsyncServerSocket's destructor is protected and cannot be invoked directly.

Definition at line 226 of file AsyncServerSocket.h.

References AsyncServerSocket(), attachEventBase(), destroy(), detachEventBase(), and setShutdownSocketSet().

Referenced by folly::TestSSLServer::init(), folly::ScopedBoundPort::ScopedBoundPort(), folly::TEST(), and TEST().

                                 {
     return std::shared_ptr<AsyncServerSocket>(
         new AsyncServerSocket(evb), Destructor());
   }

CallbackInfo* folly::AsyncServerSocket::nextCallback ( )

inlineprivate

Definition at line 814 of file AsyncServerSocket.h.

References callbackIndex_, callbacks_, and deadlock::info().

Referenced by dispatchError(), and dispatchSocket().

                                {
     CallbackInfo* info = &callbacks_[callbackIndex_];
 
     ++callbackIndex_;
     if (callbackIndex_ >= callbacks_.size()) {
       callbackIndex_ = 0;
     }
 
     return info;
   }

void folly::AsyncServerSocket::pauseAccepting ( )

Pause accepting connections.

startAccepting() may be called to resume accepting.

This method may only be called from the primary EventBase thread. If there are AcceptCallbacks being driven by other EventBase threads they may continue to receive callbacks for a short period of time after pauseAccepting() returns.

Unlike removeAcceptCallback() or destroy(), acceptStopped() will not be called on the AcceptCallback objects simply due to a temporary pause. If the server socket is later destroyed while paused, acceptStopped() will be called all of the installed AcceptCallbacks.

Definition at line 723 of file AsyncServerSocket.cpp.

References accepting_, backoffTimeout_, folly::AsyncTimeout::cancelTimeout(), folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), and sockets_.

Referenced by getAddress().

                                        {
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
   accepting_ = false;
   for (auto& handler : sockets_) {
     handler.unregisterHandler();
   }
 
   // If we were in the accept backoff state, disable the backoff timeout
   if (backoffTimeout_) {
     backoffTimeout_->cancelTimeout();
   }
 }

void folly::AsyncServerSocket::removeAcceptCallback	(	AcceptCallback *	callback,
		EventBase *	eventBase
	)

Remove an AcceptCallback.

This allows a single AcceptCallback to be removed from the round-robin pool.

Given that the accept callback is being driven by a different EventBase, the AcceptCallback may continue to be invoked for a short period of time after removeAcceptCallback() returns in this thread. Once the other EventBase thread receives the notification to stop, it will call acceptStopped() on the callback to inform it that it is fully stopped and will not receive any new sockets.

If the last accept callback is removed while the socket is accepting, the socket will implicitly pause accepting. If a callback is later added, it will resume accepting immediately, without requiring startAccepting() to be invoked.

Parameters

callback	The callback to uninstall.
eventBase	The EventBase associated with this callback. This must be the same EventBase that was used when the callback was installed with addAcceptCallback().

Definition at line 636 of file AsyncServerSocket.cpp.

References accepting_, folly::AsyncServerSocket::AcceptCallback::acceptStopped(), folly::AsyncServerSocket::CallbackInfo::callback, callbackIndex_, callbacks_, folly::AsyncServerSocket::CallbackInfo::consumer, folly::EventBase::dcheckIsInEventBaseThread(), folly::AsyncServerSocket::CallbackInfo::eventBase, eventBase_, handler(), deadlock::info(), sockets_, folly::AsyncServerSocket::RemoteAcceptor::stop(), and uint32_t.

Referenced by getAddress(), and serverSocketSanityTest().

                           {
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
 
   // Find the matching AcceptCallback.
   // We just do a simple linear search; we don't expect removeAcceptCallback()
   // to be called frequently, and we expect there to only be a small number of
   // callbacks anyway.
   std::vector<CallbackInfo>::iterator it = callbacks_.begin();
   uint32_t n = 0;
   while (true) {
     if (it == callbacks_.end()) {
       throw std::runtime_error(
           "AsyncServerSocket::removeAcceptCallback(): "
           "accept callback not found");
     }
     if (it->callback == callback &&
         (it->eventBase == eventBase || eventBase == nullptr)) {
       break;
     }
     ++it;
     ++n;
   }
 
   // Remove this callback from callbacks_.
   //
   // Do this before invoking the acceptStopped() callback, in case
   // acceptStopped() invokes one of our methods that examines callbacks_.
   //
   // Save a copy of the CallbackInfo first.
   CallbackInfo info(*it);
   callbacks_.erase(it);
   if (n < callbackIndex_) {
     // We removed an element before callbackIndex_.  Move callbackIndex_ back
     // one step, since things after n have been shifted back by 1.
     --callbackIndex_;
   } else {
     // We removed something at or after callbackIndex_.
     // If we removed the last element and callbackIndex_ was pointing at it,
     // we need to reset callbackIndex_ to 0.
     if (callbackIndex_ >= callbacks_.size()) {
       callbackIndex_ = 0;
     }
   }
 
   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.
   if (accepting_ && callbacks_.empty()) {
     for (auto& handler : sockets_) {
       handler.unregisterHandler();
     }
   }
 }

void folly::AsyncServerSocket::setAcceptRateAdjustSpeed ( double speed )

inline

Set the speed of adjusting connection accept rate.

Definition at line 576 of file AsyncServerSocket.h.

References acceptRateAdjustSpeed_, and setTosReflect().

                                               {
     acceptRateAdjustSpeed_ = speed;
   }

void folly::AsyncServerSocket::setCloseOnExec ( bool closeOnExec )

inline

Set whether or not the socket should close during exec() (FD_CLOEXEC). By default, this is enabled

Definition at line 685 of file AsyncServerSocket.h.

References closeOnExec_.

                                         {
     closeOnExec_ = closeOnExec;
   }

void folly::AsyncServerSocket::setConnectionEventCallback ( ConnectionEventCallback *const connectionEventCallback )

inline

Set the ConnectionEventCallback

Definition at line 721 of file AsyncServerSocket.h.

References connectionEventCallback_.

                                                               {
     connectionEventCallback_ = connectionEventCallback;
   }

void folly::AsyncServerSocket::setKeepAliveEnabled ( bool enabled )

inline

Set whether or not SO_KEEPALIVE should be enabled on the server socket (and thus on all subsequently-accepted connections). By default, keepalive is enabled.

Note that TCP keepalive usually only kicks in after the connection has been idle for several hours. Applications should almost always have their own, shorter idle timeout.

Definition at line 626 of file AsyncServerSocket.h.

References folly::errnoStr(), handler(), keepAliveEnabled_, folly::netops::setsockopt(), sockets_, and val.

                                          {
     keepAliveEnabled_ = enabled;
 
     for (auto& handler : sockets_) {
       if (handler.socket_ < 0) {
         continue;
       }
 
       int val = (enabled) ? 1 : 0;
       if (setsockopt(
               handler.socket_, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) !=
           0) {
         LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: %s"
                    << errnoStr(errno);
       }
     }
   }

void folly::AsyncServerSocket::setMaxAcceptAtOnce ( uint32_t numConns )

inline

Set the maximum number of connections that will be accepted each time around the event loop.

This provides a very coarse-grained way of controlling how fast the AsyncServerSocket will accept connections. If you find that when your server is overloaded AsyncServerSocket accepts connections more quickly than your code can process them, you can try lowering this number so that fewer connections will be accepted each event loop iteration.

For more explicit control over the accept rate, you can also use pauseAccepting() to temporarily pause accepting when your server is overloaded, and then use startAccepting() later to resume accepting.

Definition at line 543 of file AsyncServerSocket.h.

References maxAcceptAtOnce_.

                                              {
     maxAcceptAtOnce_ = numConns;
   }

void folly::AsyncServerSocket::setMaxNumMessagesInQueue ( uint32_t num )

inline

Set the maximum number of unprocessed messages in NotificationQueue. No new message will be sent to that NotificationQueue if there are more than such number of unprocessed messages in that queue.

Only works if called before addAcceptCallback.

Definition at line 562 of file AsyncServerSocket.h.

References maxNumMsgsInQueue_.

                                               {
     maxNumMsgsInQueue_ = num;
   }

void folly::AsyncServerSocket::setReusePortEnabled ( bool enabled )

inline

Set whether or not SO_REUSEPORT should be enabled on the server socket, allowing multiple binds to the same port

Definition at line 655 of file AsyncServerSocket.h.

References handler(), reusePortEnabled_, folly::netops::setsockopt(), SO_REUSEPORT, sockets_, folly::throwSystemError(), and val.

                                          {
     reusePortEnabled_ = enabled;
 
     for (auto& handler : sockets_) {
       if (handler.socket_ < 0) {
         continue;
       }
 
       int val = (enabled) ? 1 : 0;
       if (setsockopt(
               handler.socket_, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(val)) !=
           0) {
         LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
                    << errno;
         folly::throwSystemError(errno, "failed to bind to async server socket");
       }
     }
   }

void folly::AsyncServerSocket::setShutdownSocketSet ( const std::weak_ptr< ShutdownSocketSet > & wNewSS )

Definition at line 179 of file AsyncServerSocket.cpp.

References h, folly::ShutdownSocketSet::remove(), folly::test::shutdownSocketSet, sockets_, and wShutdownSocketSet_.

Referenced by newSocket().

                                                   {
   const auto newSS = wNewSS.lock();
   const auto shutdownSocketSet = wShutdownSocketSet_.lock();
 
   if (shutdownSocketSet == newSS) {
     return;
   }
 
   if (shutdownSocketSet) {
     for (auto& h : sockets_) {
       shutdownSocketSet->remove(h.socket_);
     }
   }
 
   if (newSS) {
     for (auto& h : sockets_) {
       newSS->add(h.socket_);
     }
   }
 
   wShutdownSocketSet_ = wNewSS;
 }

void folly::AsyncServerSocket::setTFOEnabled	(	bool	enabled,
		uint32_t	maxTFOQueueSize
	)

inline

Tries to enable TFO if the machine supports it.

Definition at line 699 of file AsyncServerSocket.h.

References tfo_, and tfoMaxQueueSize_.

                                                              {
     tfo_ = enabled;
     tfoMaxQueueSize_ = maxTFOQueueSize;
   }

void folly::AsyncServerSocket::setTosReflect ( bool enable )

Enable/Disable TOS reflection for the server socket

Enable/Disable TOS reflection for the server socket If enabled, the 'accepted' connections will reflect the TOS derived from the client's connect request

Definition at line 758 of file AsyncServerSocket.cpp.

References handler(), folly::kIsLinux, folly::netops::setsockopt(), sockets_, TCP_SAVE_SYN, folly::throwSystemError(), tosReflect_, and val.

Referenced by setAcceptRateAdjustSpeed().

                                                  {
   if (!kIsLinux || enable == false) {
     tosReflect_ = false;
     return;
   }
 
   for (auto& handler : sockets_) {
     if (handler.socket_ < 0) {
       continue;
     }
 
     int val = (enable) ? 1 : 0;
     int ret = setsockopt(
         handler.socket_, IPPROTO_TCP, TCP_SAVE_SYN, &val, sizeof(val));
 
     if (ret == 0) {
       VLOG(10) << "Enabled SYN save for socket " << handler.socket_;
     } else {
       folly::throwSystemError(errno, "failed to enable TOS reflect");
     }
   }
   tosReflect_ = true;
 }

void folly::AsyncServerSocket::setupSocket	(	int	fd,
		int	family
	)

private

Definition at line 782 of file AsyncServerSocket.cpp.

References folly::ShutdownSocketSet::add(), closeOnExec_, folly::SocketAddress::describe(), folly::errnoStr(), keepAliveEnabled_, reusePortEnabled_, folly::setCloseOnExec(), folly::SocketAddress::setFromLocalAddress(), folly::netops::setsockopt(), folly::test::shutdownSocketSet, SO_REUSEPORT, tfo_, folly::detail::tfo_enable(), tfoMaxQueueSize_, folly::throwSystemError(), folly::WARNING, and wShutdownSocketSet_.

Referenced by bind(), createSocket(), and useExistingSockets().

                                                       {
   // Put the socket in non-blocking mode
   if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
     folly::throwSystemError(errno, "failed to put socket in non-blocking mode");
   }
 
   // Set reuseaddr to avoid 2MSL delay on server restart
   int one = 1;
   if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
     // This isn't a fatal error; just log an error message and continue
     LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
   }
 
   // Set reuseport to support multiple accept threads
   int zero = 0;
   if (reusePortEnabled_ &&
       setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
     LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
                << errnoStr(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
   if (setsockopt(
           fd,
           SOL_SOCKET,
           SO_KEEPALIVE,
           (keepAliveEnabled_) ? &one : &zero,
           sizeof(int)) != 0) {
     LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: "
                << errnoStr(errno);
   }
 
   // Setup FD_CLOEXEC flag
   if (closeOnExec_ && (-1 == folly::setCloseOnExec(fd, closeOnExec_))) {
     LOG(ERROR) << "failed to set FD_CLOEXEC on async server socket: "
                << errnoStr(errno);
   }
 
   // Set TCP nodelay if available, MAC OS X Hack
   // See http://lists.danga.com/pipermail/memcached/2005-March/001240.html
 #ifndef TCP_NOPUSH
   if (family != AF_UNIX) {
     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0) {
       // This isn't a fatal error; just log an error message and continue
       LOG(ERROR) << "failed to set TCP_NODELAY on async server socket: "
                  << errnoStr(errno);
     }
   }
 #else
   (void)family; // to avoid unused parameter warning
 #endif
 
 #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);
   }
 }

bool folly::AsyncServerSocket::setZeroCopy ( bool enable )

Definition at line 360 of file AsyncServerSocket.cpp.

References getSocket(), folly::netops::setsockopt(), SO_ZEROCOPY, and val.

Referenced by getSocket().

                                                {
   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 folly::AsyncServerSocket::startAccepting ( )

virtual

Begin accepting connctions on this socket.

bind() and listen() must be called before calling startAccepting().

When a AsyncServerSocket is initially created, it will not begin accepting connections until at least one callback has been added and startAccepting() has been called. startAccepting() can also be used to resume accepting connections after a call to pauseAccepting().

If startAccepting() is called when there are no accept callbacks installed, the socket will not actually begin accepting until an accept callback is added.

This method may only be called from the primary EventBase thread.

Definition at line 704 of file AsyncServerSocket.cpp.

References accepting_, callbacks_, folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), folly::EventHandler::PERSIST, folly::EventHandler::READ, and sockets_.

Referenced by addAcceptCallback(), getAddress(), and serverSocketSanityTest().

                                        {
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
 
   accepting_ = true;
   if (callbacks_.empty()) {
     // We can't actually begin accepting if no callbacks are defined.
     // Wait until a callback is added to start accepting.
     return;
   }
 
   for (auto& handler : sockets_) {
     if (!handler.registerHandler(EventHandler::READ | EventHandler::PERSIST)) {
       throw std::runtime_error("failed to register for accept events");
     }
   }
 }

int folly::AsyncServerSocket::stopAccepting ( int shutdownFlags = -1 )

Shutdown the listen socket and notify all callbacks that accept has stopped, but don't close the socket. This invokes shutdown(2) with the supplied argument. Passing -1 will close the socket now. Otherwise, the close will be delayed until this object is destroyed.

Only use this if you have reason to pass special flags to shutdown. Otherwise just destroy the socket.

This method has no effect when a ShutdownSocketSet option is used.

Returns the result of shutdown on sockets_[n-1]

Definition at line 207 of file AsyncServerSocket.cpp.

References accepting_, backoffTimeout_, callbacks_, folly::ShutdownSocketSet::close(), folly::closeNoInt(), folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, handler(), pendingCloseSockets_, folly::shutdownNoInt(), folly::test::shutdownSocketSet, sockets_, and wShutdownSocketSet_.

Referenced by destroy(), and getAddress().

                                                       {
   int result = 0;
   for (auto& handler : sockets_) {
     VLOG(10) << "AsyncServerSocket::stopAccepting " << this << handler.socket_;
   }
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
 
   // When destroy is called, unregister and close the socket immediately.
   accepting_ = false;
 
   // 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 (const auto shutdownSocketSet = wShutdownSocketSet_.lock()) {
       shutdownSocketSet->close(handler.socket_);
     } else if (shutdownFlags >= 0) {
       result = shutdownNoInt(handler.socket_, shutdownFlags);
       pendingCloseSockets_.push_back(handler.socket_);
     } else {
       closeNoInt(handler.socket_);
     }
   }
 
   // Destroy the backoff timout.  This will cancel it if it is running.
   delete backoffTimeout_;
   backoffTimeout_ = nullptr;
 
   // Close all of the callback queues to notify them that they are being
   // destroyed.  No one should access the AsyncServerSocket any more once
   // destroy() is called.  However, clear out callbacks_ before invoking the
   // accept callbacks just in case.  This will potentially help us detect the
   // bug if one of the callbacks calls addAcceptCallback() or
   // removeAcceptCallback().
   std::vector<CallbackInfo> callbacksCopy;
   callbacks_.swap(callbacksCopy);
   for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
        it != callbacksCopy.end();
        ++it) {
     // 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;
 }

void folly::AsyncServerSocket::useExistingSocket ( int fd )

Create a AsyncServerSocket from an existing socket file descriptor.

useExistingSocket() will cause the AsyncServerSocket to take ownership of the specified file descriptor, and use it to listen for new connections. The AsyncServerSocket will close the file descriptor when it is destroyed.

useExistingSocket() must be called before bind() or listen().

The supplied file descriptor will automatically be put into non-blocking mode. The caller may have already directly called bind() and possibly listen on the file descriptor. If so the caller should skip calling the corresponding AsyncServerSocket::bind() and listen() methods.

On error a TTransportException will be thrown and the caller will retain ownership of the file descriptor.

Definition at line 327 of file AsyncServerSocket.cpp.

References useExistingSockets().

Referenced by getEventBase().

                                                 {
   useExistingSockets({fd});
 }

void folly::AsyncServerSocket::useExistingSockets ( const std::vector< int > & fds )

Definition at line 295 of file AsyncServerSocket.cpp.

References folly::EventBase::dcheckIsInEventBaseThread(), eventBase_, folly::SocketAddress::getFamily(), noTransparentTls_, folly::SocketAddress::setFromLocalAddress(), folly::netops::setsockopt(), setupSocket(), and sockets_.

Referenced by getEventBase(), and useExistingSocket().

                                                                     {
   if (eventBase_) {
     eventBase_->dcheckIsInEventBaseThread();
   }
 
   if (sockets_.size() > 0) {
     throw std::invalid_argument(
         "cannot call useExistingSocket() on a "
         "AsyncServerSocket that already has a socket");
   }
 
   for (auto fd : fds) {
     // Set addressFamily_ from this socket.
     // Note that the socket may not have been bound yet, but
     // setFromLocalAddress() will still work and get the correct address family.
     // We will update addressFamily_ again anyway if bind() is called later.
     SocketAddress address;
     address.setFromLocalAddress(fd);
 
 #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);
   }
 }