/*
- * 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.
#include <folly/io/async/AsyncServerSocket.h>
#include <folly/FileUtil.h>
+#include <folly/Portability.h>
#include <folly/SocketAddress.h>
#include <folly/String.h>
#include <folly/detail/SocketFastOpen.h>
#include <string.h>
#include <sys/types.h>
+namespace fsp = folly::portability::sockets;
+
namespace folly {
const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce;
}
void AsyncServerSocket::RemoteAcceptor::messageAvailable(
- QueueMessage&& msg) {
-
+ QueueMessage&& msg) noexcept {
switch (msg.type) {
case MessageType::MSG_NEW_CONN:
{
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(); }
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;
for (std::vector<CallbackInfo>::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;
void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
assert(eventBase_ == nullptr);
- assert(eventBase->isInEventBaseThread());
+ eventBase->dcheckIsInEventBaseThread();
eventBase_ = eventBase;
for (auto& handler : sockets_) {
void AsyncServerSocket::detachEventBase() {
assert(eventBase_ != nullptr);
- assert(eventBase_->isInEventBaseThread());
+ eventBase_->dcheckIsInEventBaseThread();
assert(!accepting_);
eventBase_ = nullptr;
}
void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+ if (eventBase_) {
+ eventBase_->dcheckIsInEventBaseThread();
+ }
if (sockets_.size() > 0) {
throw std::invalid_argument(
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);
sockaddr_storage addrStorage;
address.getAddress(&addrStorage);
sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
- if (::bind(fd, saddr, address.getActualSize()) != 0) {
+
+ if (fsp::bind(fd, saddr, address.getActualSize()) != 0) {
if (!isExistingSocket) {
closeNoInt(fd);
}
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());
}
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.
}
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");
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;
&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 ",
" 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;
// - 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);
}
}
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
}
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");
}
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
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()
}
}
- 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.
}
void AsyncServerSocket::startAccepting() {
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+ if (eventBase_) {
+ eventBase_->dcheckIsInEventBaseThread();
+ }
accepting_ = true;
if (callbacks_.empty()) {
}
void AsyncServerSocket::pauseAccepting() {
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+ if (eventBase_) {
+ eventBase_->dcheckIsInEventBaseThread();
+ }
accepting_ = false;
for (auto& handler : sockets_) {
handler. unregisterHandler();
}
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");
}
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.
// 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()) {
}
}
-
-
-} // folly
+} // namespace folly
projects / folly.git / blobdiff