X-Git-Url: http://plrg.eecs.uci.edu/git/?a=blobdiff_plain;f=folly%2Fio%2Fasync%2FAsyncServerSocket.cpp;h=985f3d3fefea0a230bb2d1cc3b12c85c8864c6e0;hb=dc1c3dcc04baa70029d16351d4424af4a7ef4385;hp=fccf662d745c4c52d462788de1faa6a7a1aa7f43;hpb=0d8402e0320e90b87d312967ff039c19417ca4f1;p=folly.git diff --git a/folly/io/async/AsyncServerSocket.cpp b/folly/io/async/AsyncServerSocket.cpp index fccf662d..985f3d3f 100644 --- a/folly/io/async/AsyncServerSocket.cpp +++ b/folly/io/async/AsyncServerSocket.cpp @@ -1,5 +1,5 @@ /* - * Copyright 2016 Facebook, Inc. + * Copyright 2017 Facebook, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,17 +21,21 @@ #include #include +#include #include +#include +#include #include #include +#include +#include +#include #include -#include -#include #include -#include #include -#include + +namespace fsp = folly::portability::sockets; namespace folly { @@ -86,8 +90,7 @@ void AsyncServerSocket::RemoteAcceptor::stop( } void AsyncServerSocket::RemoteAcceptor::messageAvailable( - QueueMessage&& msg) { - + QueueMessage&& msg) noexcept { switch (msg.type) { case MessageType::MSG_NEW_CONN: { @@ -122,7 +125,7 @@ class AsyncServerSocket::BackoffTimeout : public AsyncTimeout { public: // Disallow copy, move, and default constructors. BackoffTimeout(BackoffTimeout&&) = delete; - BackoffTimeout(AsyncServerSocket* socket) + explicit BackoffTimeout(AsyncServerSocket* socket) : AsyncTimeout(socket->getEventBase()), socket_(socket) {} void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); } @@ -179,7 +182,9 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { VLOG(10) << "AsyncServerSocket::stopAccepting " << this << handler.socket_; } - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // When destroy is called, unregister and close the socket immediately. accepting_ = false; @@ -216,7 +221,14 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { for (std::vector::iterator it = callbacksCopy.begin(); it != callbacksCopy.end(); ++it) { - it->consumer->stop(it->eventBase, it->callback); + // consumer may not be set if we are running in primary event base + if (it->consumer) { + DCHECK(it->eventBase); + it->consumer->stop(it->eventBase, it->callback); + } else { + DCHECK(it->callback); + it->callback->acceptStopped(); + } } return result; @@ -234,7 +246,7 @@ void AsyncServerSocket::destroy() { void AsyncServerSocket::attachEventBase(EventBase *eventBase) { assert(eventBase_ == nullptr); - assert(eventBase->isInEventBaseThread()); + eventBase->dcheckIsInEventBaseThread(); eventBase_ = eventBase; for (auto& handler : sockets_) { @@ -244,7 +256,7 @@ void AsyncServerSocket::attachEventBase(EventBase *eventBase) { void AsyncServerSocket::detachEventBase() { assert(eventBase_ != nullptr); - assert(eventBase_->isInEventBaseThread()); + eventBase_->dcheckIsInEventBaseThread(); assert(!accepting_); eventBase_ = nullptr; @@ -254,7 +266,9 @@ void AsyncServerSocket::detachEventBase() { } void AsyncServerSocket::useExistingSockets(const std::vector& fds) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } if (sockets_.size() > 0) { throw std::invalid_argument( @@ -270,6 +284,13 @@ void AsyncServerSocket::useExistingSockets(const std::vector& fds) { 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); @@ -287,7 +308,8 @@ void AsyncServerSocket::bindSocket( sockaddr_storage addrStorage; address.getAddress(&addrStorage); sockaddr* saddr = reinterpret_cast(&addrStorage); - if (::bind(fd, saddr, address.getActualSize()) != 0) { + + if (fsp::bind(fd, saddr, address.getActualSize()) != 0) { if (!isExistingSocket) { closeNoInt(fd); } @@ -296,6 +318,13 @@ void AsyncServerSocket::bindSocket( address.describe()); } +#if __linux__ + if (noTransparentTls_) { + // Ignore return value, errors are ok + setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0); + } +#endif + // If we just created this socket, update the EventHandler and set socket_ if (!isExistingSocket) { sockets_.emplace_back(eventBase_, fd, this, address.getFamily()); @@ -303,7 +332,9 @@ void AsyncServerSocket::bindSocket( } void AsyncServerSocket::bind(const SocketAddress& address) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // useExistingSocket() may have been called to initialize socket_ already. // However, in the normal case we need to create a new socket now. @@ -351,16 +382,19 @@ void AsyncServerSocket::bind( } void AsyncServerSocket::bind(uint16_t port) { - struct addrinfo hints, *res, *res0; + struct addrinfo hints, *res0; char sport[sizeof("65536")]; memset(&hints, 0, sizeof(hints)); hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM; - hints.ai_flags = AI_PASSIVE; + hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV; snprintf(sport, sizeof(sport), "%u", port); - if (getaddrinfo(nullptr, sport, &hints, &res0)) { + // On Windows the value we need to pass to bind to all available + // addresses is an empty string. Everywhere else, it's nullptr. + constexpr const char* kWildcardNode = kIsWindows ? "" : nullptr; + if (getaddrinfo(kWildcardNode, sport, &hints, &res0)) { throw std::invalid_argument( "Attempted to bind address to socket with " "bad getaddrinfo"); @@ -369,7 +403,7 @@ void AsyncServerSocket::bind(uint16_t port) { SCOPE_EXIT { freeaddrinfo(res0); }; auto setupAddress = [&] (struct addrinfo* res) { - int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + int s = fsp::socket(res->ai_family, res->ai_socktype, res->ai_protocol); // IPv6/IPv4 may not be supported by the kernel if (s < 0 && errno == EAFNOSUPPORT) { return; @@ -389,13 +423,8 @@ void AsyncServerSocket::bind(uint16_t port) { &v6only, sizeof(v6only))); } - SocketAddress address; - address.setFromLocalAddress(s); - - sockets_.emplace_back(eventBase_, s, this, address.getFamily()); - // Bind to the socket - if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) { + if (fsp::bind(s, res->ai_addr, socklen_t(res->ai_addrlen)) != 0) { folly::throwSystemError( errno, "failed to bind to async server socket for port ", @@ -403,6 +432,18 @@ void AsyncServerSocket::bind(uint16_t port) { " family ", SocketAddress::getFamilyNameFrom(res->ai_addr, "")); } + +#if __linux__ + if (noTransparentTls_) { + // Ignore return value, errors are ok + setsockopt(s, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0); + } +#endif + + SocketAddress address; + address.setFromLocalAddress(s); + + sockets_.emplace_back(eventBase_, s, this, address.getFamily()); }; const int kNumTries = 25; @@ -413,7 +454,7 @@ void AsyncServerSocket::bind(uint16_t port) { // - 0.0.0.0 (IPv4-only) // - :: (IPv6+IPv4) in this order // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981 - for (res = res0; res; res = res->ai_next) { + for (struct addrinfo* res = res0; res; res = res->ai_next) { if (res->ai_family == AF_INET6) { setupAddress(res); } @@ -430,12 +471,12 @@ void AsyncServerSocket::bind(uint16_t port) { } try { - for (res = res0; res; res = res->ai_next) { + for (struct addrinfo* res = res0; res; res = res->ai_next) { if (res->ai_family != AF_INET6) { setupAddress(res); } } - } catch (const std::system_error& e) { + } 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 @@ -470,11 +511,13 @@ void AsyncServerSocket::bind(uint16_t port) { } void AsyncServerSocket::listen(int backlog) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // Start listening for (auto& handler : sockets_) { - if (::listen(handler.socket_, backlog) == -1) { + if (fsp::listen(handler.socket_, backlog) == -1) { folly::throwSystemError(errno, "failed to listen on async server socket"); } @@ -504,18 +547,31 @@ std::vector AsyncServerSocket::getAddresses() void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback, EventBase *eventBase, uint32_t maxAtOnce) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // If this is the first accept callback and we are supposed to be accepting, // start accepting once the callback is installed. bool runStartAccepting = accepting_ && callbacks_.empty(); + callbacks_.emplace_back(callback, eventBase); + + SCOPE_SUCCESS { + // If this is the first accept callback and we are supposed to be accepting, + // start accepting. + if (runStartAccepting) { + startAccepting(); + } + }; + if (!eventBase) { - eventBase = eventBase_; // Run in AsyncServerSocket's eventbase + // Run in AsyncServerSocket's eventbase; notify that we are + // starting to accept connections + callback->acceptStarted(); + return; } - callbacks_.emplace_back(callback, eventBase); - // Start the remote acceptor. // // It would be nice if we could avoid starting the remote acceptor if @@ -535,17 +591,13 @@ void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback, throw; } callbacks_.back().consumer = acceptor; - - // If this is the first accept callback and we are supposed to be accepting, - // start accepting. - if (runStartAccepting) { - startAccepting(); - } } void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, EventBase *eventBase) { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } // Find the matching AcceptCallback. // We just do a simple linear search; we don't expect removeAcceptCallback() @@ -587,7 +639,16 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, } } - info.consumer->stop(info.eventBase, info.callback); + if (info.consumer) { + // consumer could be nullptr is we run callbacks in primary event + // base + DCHECK(info.eventBase); + info.consumer->stop(info.eventBase, info.callback); + } else { + // callback invoked in the primary event base, just call directly + DCHECK(info.callback); + callback->acceptStopped(); + } // If we are supposed to be accepting but the last accept callback // was removed, unregister for events until a callback is added. @@ -599,7 +660,9 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback, } void AsyncServerSocket::startAccepting() { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } accepting_ = true; if (callbacks_.empty()) { @@ -617,7 +680,9 @@ void AsyncServerSocket::startAccepting() { } void AsyncServerSocket::pauseAccepting() { - assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread()); + if (eventBase_) { + eventBase_->dcheckIsInEventBaseThread(); + } accepting_ = false; for (auto& handler : sockets_) { handler. unregisterHandler(); @@ -630,7 +695,7 @@ void AsyncServerSocket::pauseAccepting() { } int AsyncServerSocket::createSocket(int family) { - int fd = socket(family, SOCK_STREAM, 0); + int fd = fsp::socket(family, SOCK_STREAM, 0); if (fd == -1) { folly::throwSystemError(errno, "error creating async server socket"); } @@ -701,6 +766,14 @@ void AsyncServerSocket::setupSocket(int fd, int family) { } #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 (shutdownSocketSet_) { shutdownSocketSet_->add(fd); } @@ -917,7 +990,7 @@ void AsyncServerSocket::enterBackoff() { if (backoffTimeout_ == nullptr) { try { backoffTimeout_ = new BackoffTimeout(this); - } catch (const std::bad_alloc& ex) { + } catch (const std::bad_alloc&) { // Man, we couldn't even allocate the timer to re-enable accepts. // We must be in pretty bad shape. Don't pause accepting for now, // since we won't be able to re-enable ourselves later. @@ -966,7 +1039,8 @@ void AsyncServerSocket::backoffTimeoutExpired() { // the backoff timeout. assert(accepting_); // We can't be detached from the EventBase without being paused - assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread()); + assert(eventBase_ != nullptr); + eventBase_->dcheckIsInEventBaseThread(); // If all of the callbacks were removed, we shouldn't re-enable accepts if (callbacks_.empty()) {