Commit d280bbd5 authored by Brad Kim's avatar Brad Kim

Merge commit '64f2f273' into fix-configure-pthread-atfork

parents 462e13b1 64f2f273
...@@ -138,7 +138,6 @@ nobase_follyinclude_HEADERS = \ ...@@ -138,7 +138,6 @@ nobase_follyinclude_HEADERS = \
io/async/AsyncUDPServerSocket.h \ io/async/AsyncUDPServerSocket.h \
io/async/AsyncUDPSocket.h \ io/async/AsyncUDPSocket.h \
io/async/AsyncServerSocket.h \ io/async/AsyncServerSocket.h \
io/async/AsyncSSLServerSocket.h \
io/async/AsyncSocket.h \ io/async/AsyncSocket.h \
io/async/AsyncSSLSocket.h \ io/async/AsyncSSLSocket.h \
io/async/AsyncSocketException.h \ io/async/AsyncSocketException.h \
...@@ -308,7 +307,6 @@ libfolly_la_SOURCES = \ ...@@ -308,7 +307,6 @@ libfolly_la_SOURCES = \
io/async/AsyncTimeout.cpp \ io/async/AsyncTimeout.cpp \
io/async/AsyncUDPSocket.cpp \ io/async/AsyncUDPSocket.cpp \
io/async/AsyncServerSocket.cpp \ io/async/AsyncServerSocket.cpp \
io/async/AsyncSSLServerSocket.cpp \
io/async/AsyncSocket.cpp \ io/async/AsyncSocket.cpp \
io/async/AsyncSSLSocket.cpp \ io/async/AsyncSSLSocket.cpp \
io/async/EventBase.cpp \ io/async/EventBase.cpp \
......
...@@ -20,6 +20,12 @@ ...@@ -20,6 +20,12 @@
namespace folly { namespace folly {
namespace {
static constexpr std::chrono::seconds kDestroyWaitTime{5};
}
SingletonVault::~SingletonVault() { destroyInstances(); } SingletonVault::~SingletonVault() { destroyInstances(); }
void SingletonVault::destroyInstances() { void SingletonVault::destroyInstances() {
...@@ -57,14 +63,19 @@ void SingletonVault::destroyInstances() { ...@@ -57,14 +63,19 @@ void SingletonVault::destroyInstances() {
void SingletonVault::destroyInstance(SingletonMap::iterator entry_it) { void SingletonVault::destroyInstance(SingletonMap::iterator entry_it) {
const auto& type = entry_it->first; const auto& type = entry_it->first;
auto& entry = *(entry_it->second); auto& entry = *(entry_it->second);
if (entry.instance.use_count() > 1) {
LOG(ERROR) << "Singleton of type " << type.name() << " has a living "
<< "reference at destroyInstances time; beware! Raw pointer "
<< "is " << entry.instance.get() << " with use_count of "
<< entry.instance.use_count();
}
entry.state = detail::SingletonEntryState::Dead; entry.state = detail::SingletonEntryState::Dead;
entry.instance.reset(); entry.instance.reset();
auto wait_result = entry.destroy_baton->timed_wait(
std::chrono::steady_clock::now() + kDestroyWaitTime);
if (!wait_result) {
LOG(ERROR) << "Singleton of type " << type.prettyName() << " has a living "
<< "reference at destroyInstances time; beware! Raw pointer "
<< "is " << entry.instance_ptr << ". It is very likely that "
<< "some other singleton is holding a shared_ptr to it. Make "
<< "sure dependencies between these singletons are properly "
<< "defined.";
}
} }
void SingletonVault::reenableInstances() { void SingletonVault::reenableInstances() {
......
...@@ -92,10 +92,12 @@ ...@@ -92,10 +92,12 @@
// should call reenableInstances. // should call reenableInstances.
#pragma once #pragma once
#include <folly/Baton.h>
#include <folly/Exception.h> #include <folly/Exception.h>
#include <folly/Hash.h> #include <folly/Hash.h>
#include <folly/Memory.h> #include <folly/Memory.h>
#include <folly/RWSpinLock.h> #include <folly/RWSpinLock.h>
#include <folly/Demangle.h>
#include <folly/io/async/Request.h> #include <folly/io/async/Request.h>
#include <algorithm> #include <algorithm>
...@@ -161,13 +163,13 @@ class TypeDescriptor { ...@@ -161,13 +163,13 @@ class TypeDescriptor {
return *this; return *this;
} }
std::string name() const { std::string prettyName() const {
std::string ret = ti_.name(); auto ret = demangle(ti_.name());
if (tag_ti_ != std::type_index(typeid(DefaultTag))) { if (tag_ti_ != std::type_index(typeid(DefaultTag))) {
ret += "/"; ret += "/";
ret += tag_ti_.name(); ret += demangle(tag_ti_.name());
} }
return ret; return ret.toStdString();
} }
friend class TypeDescriptorHasher; friend class TypeDescriptorHasher;
...@@ -223,6 +225,8 @@ struct SingletonEntry { ...@@ -223,6 +225,8 @@ struct SingletonEntry {
// safe to read it from different threads w/o synchronization if we know // safe to read it from different threads w/o synchronization if we know
// that state is set to Living // that state is set to Living
std::weak_ptr<void> instance_weak; std::weak_ptr<void> instance_weak;
// Time we wait on destroy_baton after releasing Singleton shared_ptr.
std::shared_ptr<folly::Baton<>> destroy_baton;
void* instance_ptr = nullptr; void* instance_ptr = nullptr;
CreateFunc create = nullptr; CreateFunc create = nullptr;
TeardownFunc teardown = nullptr; TeardownFunc teardown = nullptr;
...@@ -433,7 +437,7 @@ class SingletonVault { ...@@ -433,7 +437,7 @@ class SingletonVault {
auto it = singletons_.find(type); auto it = singletons_.find(type);
if (it == singletons_.end()) { if (it == singletons_.end()) {
throw std::out_of_range(std::string("non-existent singleton: ") + throw std::out_of_range(std::string("non-existent singleton: ") +
type.name()); type.prettyName());
} }
return it->second.get(); return it->second.get();
...@@ -454,7 +458,7 @@ class SingletonVault { ...@@ -454,7 +458,7 @@ class SingletonVault {
// it if it was set by current thread anyways. // it if it was set by current thread anyways.
if (entry->creating_thread == std::this_thread::get_id()) { if (entry->creating_thread == std::this_thread::get_id()) {
throw std::out_of_range(std::string("circular singleton dependency: ") + throw std::out_of_range(std::string("circular singleton dependency: ") +
type.name()); type.prettyName());
} }
std::lock_guard<std::mutex> entry_lock(entry->mutex); std::lock_guard<std::mutex> entry_lock(entry->mutex);
...@@ -471,8 +475,16 @@ class SingletonVault { ...@@ -471,8 +475,16 @@ class SingletonVault {
return entry; return entry;
} }
auto destroy_baton = std::make_shared<folly::Baton<>>();
auto teardown = entry->teardown;
// Can't use make_shared -- no support for a custom deleter, sadly. // Can't use make_shared -- no support for a custom deleter, sadly.
auto instance = std::shared_ptr<void>(entry->create(), entry->teardown); auto instance = std::shared_ptr<void>(
entry->create(),
[destroy_baton, teardown](void* instance_ptr) mutable {
teardown(instance_ptr);
destroy_baton->post();
});
// We should schedule destroyInstances() only after the singleton was // We should schedule destroyInstances() only after the singleton was
// created. This will ensure it will be destroyed before singletons, // created. This will ensure it will be destroyed before singletons,
...@@ -484,6 +496,7 @@ class SingletonVault { ...@@ -484,6 +496,7 @@ class SingletonVault {
entry->instance_weak = instance; entry->instance_weak = instance;
entry->instance_ptr = instance.get(); entry->instance_ptr = instance.get();
entry->creating_thread = std::thread::id(); entry->creating_thread = std::thread::id();
entry->destroy_baton = std::move(destroy_baton);
// This has to be the last step, because once state is Living other threads // This has to be the last step, because once state is Living other threads
// may access instance and instance_weak w/o synchronization. // may access instance and instance_weak w/o synchronization.
......
...@@ -241,8 +241,14 @@ TEST(Singleton, SharedPtrUsage) { ...@@ -241,8 +241,14 @@ TEST(Singleton, SharedPtrUsage) {
EXPECT_NE(locked.get(), shared_s1.get()); EXPECT_NE(locked.get(), shared_s1.get());
} }
LOG(ERROR) << "The following log message regarding ref counts is expected"; LOG(ERROR) << "The following log message regarding shared_ptr is expected";
vault.destroyInstances(); {
auto start_time = std::chrono::steady_clock::now();
vault.destroyInstances();
auto duration = std::chrono::steady_clock::now() - start_time;
EXPECT_TRUE(duration > std::chrono::seconds{4} &&
duration < std::chrono::seconds{6});
}
EXPECT_EQ(vault.registeredSingletonCount(), 3); EXPECT_EQ(vault.registeredSingletonCount(), 3);
EXPECT_EQ(vault.livingSingletonCount(), 0); EXPECT_EQ(vault.livingSingletonCount(), 0);
...@@ -270,6 +276,23 @@ TEST(Singleton, SharedPtrUsage) { ...@@ -270,6 +276,23 @@ TEST(Singleton, SharedPtrUsage) {
// Singleton should be re-created only after reenableInstances() was called. // Singleton should be re-created only after reenableInstances() was called.
Watchdog* new_s1 = Singleton<Watchdog>::get(&vault); Watchdog* new_s1 = Singleton<Watchdog>::get(&vault);
EXPECT_NE(new_s1->serial_number, old_serial); EXPECT_NE(new_s1->serial_number, old_serial);
auto new_s1_weak = Singleton<Watchdog>::get_weak(&vault);
auto new_s1_shared = new_s1_weak.lock();
std::thread t([new_s1_shared]() mutable {
std::this_thread::sleep_for(std::chrono::seconds{2});
new_s1_shared.reset();
});
new_s1_shared.reset();
{
auto start_time = std::chrono::steady_clock::now();
vault.destroyInstances();
auto duration = std::chrono::steady_clock::now() - start_time;
EXPECT_TRUE(duration > std::chrono::seconds{1} &&
duration < std::chrono::seconds{3});
}
EXPECT_TRUE(new_s1_weak.expired());
t.join();
} }
// Some classes to test singleton dependencies. NeedySingleton has a // Some classes to test singleton dependencies. NeedySingleton has a
......
...@@ -42,6 +42,36 @@ Future<T>& Future<T>::operator=(Future<T>&& other) { ...@@ -42,6 +42,36 @@ Future<T>& Future<T>::operator=(Future<T>&& other) {
return *this; return *this;
} }
template <class T>
template <class F>
Future<T>::Future(
const typename std::enable_if<!std::is_void<F>::value, F>::type& val)
: core_(nullptr) {
Promise<F> p;
p.setValue(val);
*this = p.getFuture();
}
template <class T>
template <class F>
Future<T>::Future(
typename std::enable_if<!std::is_void<F>::value, F>::type&& val)
: core_(nullptr) {
Promise<F> p;
p.setValue(std::forward<F>(val));
*this = p.getFuture();
}
template <>
template <class F,
typename std::enable_if<std::is_void<F>::value, int>::type>
Future<void>::Future() : core_(nullptr) {
Promise<void> p;
p.setValue();
*this = p.getFuture();
}
template <class T> template <class T>
Future<T>::~Future() { Future<T>::~Future() {
detach(); detach();
......
...@@ -181,6 +181,19 @@ class Future { ...@@ -181,6 +181,19 @@ class Future {
Future(Future&&) noexcept; Future(Future&&) noexcept;
Future& operator=(Future&&); Future& operator=(Future&&);
// makeFuture
template <class F = T>
/* implicit */
Future(const typename std::enable_if<!std::is_void<F>::value, F>::type& val);
template <class F = T>
/* implicit */
Future(typename std::enable_if<!std::is_void<F>::value, F>::type&& val);
template <class F = T,
typename std::enable_if<std::is_void<F>::value, int>::type = 0>
Future();
~Future(); ~Future();
/** Return the reference to result. Should not be called if !isReady(). /** Return the reference to result. Should not be called if !isReady().
......
...@@ -1311,3 +1311,18 @@ TEST(Future, CircularDependencySharedPtrSelfReset) { ...@@ -1311,3 +1311,18 @@ TEST(Future, CircularDependencySharedPtrSelfReset) {
promise.fulfil([]{return 1l;}); promise.fulfil([]{return 1l;});
} }
TEST(Future, Constructor) {
auto f1 = []() -> Future<int> { return Future<int>(3); }();
EXPECT_EQ(f1.value(), 3);
auto f2 = []() -> Future<void> { return Future<void>(); }();
EXPECT_NO_THROW(f2.value());
}
TEST(Future, ImplicitConstructor) {
auto f1 = []() -> Future<int> { return 3; }();
EXPECT_EQ(f1.value(), 3);
// Unfortunately, the C++ standard does not allow the
// following implicit conversion to work:
//auto f2 = []() -> Future<void> { }();
}
...@@ -35,6 +35,7 @@ static vector<fbstring> testStrVector ...@@ -35,6 +35,7 @@ static vector<fbstring> testStrVector
= seq(1, testSize.load()) = seq(1, testSize.load())
| eachTo<fbstring>() | eachTo<fbstring>()
| as<vector>(); | as<vector>();
static auto testFileContent = from(testStrVector) | unsplit('\n');
const char* const kLine = "The quick brown fox jumped over the lazy dog.\n"; const char* const kLine = "The quick brown fox jumped over the lazy dog.\n";
const size_t kLineCount = 10000; const size_t kLineCount = 10000;
...@@ -212,6 +213,20 @@ BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 2000) ...@@ -212,6 +213,20 @@ BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 2000)
BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 4000) BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 4000)
BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 8000) BENCHMARK_RELATIVE_PARAM(StringUnsplit_Gen, 8000)
BENCHMARK_DRAW_LINE()
void Lines_Gen(size_t iters, int joinSize) {
size_t s = 0;
StringPiece content = testFileContent;
for (size_t i = 0; i < iters; ++i) {
s += lines(content.subpiece(0, joinSize)) | take(100) | count;
}
folly::doNotOptimizeAway(s);
}
BENCHMARK_PARAM(Lines_Gen, 1e3)
BENCHMARK_RELATIVE_PARAM(Lines_Gen, 2e3)
BENCHMARK_RELATIVE_PARAM(Lines_Gen, 3e3)
BENCHMARK_DRAW_LINE() BENCHMARK_DRAW_LINE()
fbstring records fbstring records
......
/*
* Copyright 2014 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/io/async/AsyncSSLServerSocket.h>
#include <folly/io/async/AsyncSSLSocket.h>
#include <folly/SocketAddress.h>
using std::shared_ptr;
namespace folly {
AsyncSSLServerSocket::AsyncSSLServerSocket(
const shared_ptr<SSLContext>& ctx,
EventBase* eventBase)
: eventBase_(eventBase)
, serverSocket_(new AsyncServerSocket(eventBase))
, ctx_(ctx)
, sslCallback_(nullptr) {
}
AsyncSSLServerSocket::~AsyncSSLServerSocket() {
}
void AsyncSSLServerSocket::destroy() {
// Stop accepting on the underlying socket as soon as destroy is called
if (sslCallback_ != nullptr) {
serverSocket_->pauseAccepting();
serverSocket_->removeAcceptCallback(this, nullptr);
}
serverSocket_->destroy();
serverSocket_ = nullptr;
sslCallback_ = nullptr;
DelayedDestruction::destroy();
}
void AsyncSSLServerSocket::setSSLAcceptCallback(SSLAcceptCallback* callback) {
SSLAcceptCallback *oldCallback = sslCallback_;
sslCallback_ = callback;
if (callback != nullptr && oldCallback == nullptr) {
serverSocket_->addAcceptCallback(this, nullptr);
serverSocket_->startAccepting();
} else if (callback == nullptr && oldCallback != nullptr) {
serverSocket_->removeAcceptCallback(this, nullptr);
serverSocket_->pauseAccepting();
}
}
void AsyncSSLServerSocket::attachEventBase(EventBase* eventBase) {
assert(sslCallback_ == nullptr);
eventBase_ = eventBase;
serverSocket_->attachEventBase(eventBase);
}
void AsyncSSLServerSocket::detachEventBase() {
serverSocket_->detachEventBase();
eventBase_ = nullptr;
}
void
AsyncSSLServerSocket::connectionAccepted(
int fd,
const folly::SocketAddress& clientAddr) noexcept {
shared_ptr<AsyncSSLSocket> sslSock;
try {
// Create a AsyncSSLSocket object with the fd. The socket should be
// added to the event base and in the state of accepting SSL connection.
sslSock = AsyncSSLSocket::newSocket(ctx_, eventBase_, fd);
} catch (const std::exception &e) {
LOG(ERROR) << "Exception %s caught while creating a AsyncSSLSocket "
"object with socket " << e.what() << fd;
::close(fd);
sslCallback_->acceptError(e);
return;
}
// TODO: Perform the SSL handshake before invoking the callback
sslCallback_->connectionAccepted(sslSock);
}
void AsyncSSLServerSocket::acceptError(const std::exception& ex)
noexcept {
LOG(ERROR) << "AsyncSSLServerSocket accept error: " << ex.what();
sslCallback_->acceptError(ex);
}
} // namespace
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#pragma once
#include <folly/io/async/SSLContext.h>
#include <folly/io/async/AsyncServerSocket.h>
namespace folly {
class SocketAddress;
class AsyncSSLSocket;
class AsyncSSLServerSocket : public DelayedDestruction,
private AsyncServerSocket::AcceptCallback {
public:
class SSLAcceptCallback {
public:
virtual ~SSLAcceptCallback() {}
/**
* connectionAccepted() is called whenever a new client connection is
* received.
*
* The SSLAcceptCallback will remain installed after connectionAccepted()
* returns.
*
* @param sock The newly accepted client socket. The
* SSLAcceptCallback
* assumes ownership of this socket, and is responsible
* for closing it when done.
*/
virtual void connectionAccepted(
const std::shared_ptr<AsyncSSLSocket> &sock)
noexcept = 0;
/**
* acceptError() is called if an error occurs while accepting.
*
* The SSLAcceptCallback will remain installed even after an accept error.
* If the callback wants to uninstall itself and stop trying to accept new
* connections, it must explicit call setAcceptCallback(nullptr).
*
* @param ex An exception representing the error.
*/
virtual void acceptError(const std::exception& ex) noexcept = 0;
};
/**
* Create a new TAsyncSSLServerSocket with the specified EventBase.
*
* @param 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. All TAsyncSSLSocket objects accepted by
* this server socket will be attached to this EventBase
* when they are created.
*/
explicit AsyncSSLServerSocket(
const std::shared_ptr<folly::SSLContext>& ctx,
EventBase* eventBase = nullptr);
/**
* Destroy the socket.
*
* 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 TAsyncSSLServerSocket while it is invoking a callback.
*/
virtual void destroy();
virtual void bind(const folly::SocketAddress& address) {
serverSocket_->bind(address);
}
virtual void bind(uint16_t port) {
serverSocket_->bind(port);
}
void getAddress(folly::SocketAddress* addressReturn) {
serverSocket_->getAddress(addressReturn);
}
virtual void listen(int backlog) {
serverSocket_->listen(backlog);
}
/**
* Helper function to create a shared_ptr<TAsyncSSLServerSocket>.
*
* This passes in the correct destructor object, since TAsyncSSLServerSocket's
* destructor is protected and cannot be invoked directly.
*/
static std::shared_ptr<AsyncSSLServerSocket> newSocket(
const std::shared_ptr<folly::SSLContext>& ctx,
EventBase* evb) {
return std::shared_ptr<AsyncSSLServerSocket>(
new AsyncSSLServerSocket(ctx, evb),
Destructor());
}
/**
* Set the accept callback.
*
* This method may only be invoked from the EventBase's loop thread.
*
* @param callback The callback to invoke when a new socket
* connection is accepted and a new TAsyncSSLSocket is
* created.
*
* Throws TTransportException on error.
*/
void setSSLAcceptCallback(SSLAcceptCallback* callback);
SSLAcceptCallback *getSSLAcceptCallback() const {
return sslCallback_;
}
void attachEventBase(EventBase* eventBase);
void detachEventBase();
/**
* Returns the EventBase that the handler is currently attached to.
*/
EventBase* getEventBase() const {
return eventBase_;
}
protected:
/**
* Protected destructor.
*
* Invoke destroy() instead to destroy the TAsyncSSLServerSocket.
*/
virtual ~AsyncSSLServerSocket();
protected:
virtual void connectionAccepted(int fd,
const folly::SocketAddress& clientAddr)
noexcept;
virtual void acceptError(const std::exception& ex) noexcept;
EventBase* eventBase_;
AsyncServerSocket* serverSocket_;
// SSL context
std::shared_ptr<folly::SSLContext> ctx_;
// The accept callback
SSLAcceptCallback* sslCallback_;
};
} // namespace
...@@ -20,17 +20,18 @@ ...@@ -20,17 +20,18 @@
#include <folly/io/async/AsyncServerSocket.h> #include <folly/io/async/AsyncServerSocket.h>
#include <folly/FileUtil.h>
#include <folly/SocketAddress.h>
#include <folly/io/async/EventBase.h> #include <folly/io/async/EventBase.h>
#include <folly/io/async/NotificationQueue.h> #include <folly/io/async/NotificationQueue.h>
#include <folly/SocketAddress.h>
#include <errno.h> #include <errno.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h> #include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/tcp.h> #include <netinet/tcp.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
namespace folly { namespace folly {
...@@ -185,10 +186,10 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { ...@@ -185,10 +186,10 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) {
if (shutdownSocketSet_) { if (shutdownSocketSet_) {
shutdownSocketSet_->close(handler.socket_); shutdownSocketSet_->close(handler.socket_);
} else if (shutdownFlags >= 0) { } else if (shutdownFlags >= 0) {
result = ::shutdown(handler.socket_, shutdownFlags); result = shutdownNoInt(handler.socket_, shutdownFlags);
pendingCloseSockets_.push_back(handler.socket_); pendingCloseSockets_.push_back(handler.socket_);
} else { } else {
::close(handler.socket_); closeNoInt(handler.socket_);
} }
} }
sockets_.clear(); sockets_.clear();
...@@ -216,8 +217,8 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) { ...@@ -216,8 +217,8 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) {
void AsyncServerSocket::destroy() { void AsyncServerSocket::destroy() {
stopAccepting(); stopAccepting();
for (auto s: pendingCloseSockets_) { for (auto s : pendingCloseSockets_) {
::close(s); closeNoInt(s);
} }
// Then call DelayedDestruction::destroy() to take care of // Then call DelayedDestruction::destroy() to take care of
// whether or not we need immediate or delayed destruction // whether or not we need immediate or delayed destruction
...@@ -282,7 +283,7 @@ void AsyncServerSocket::bindSocket( ...@@ -282,7 +283,7 @@ void AsyncServerSocket::bindSocket(
sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage); sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
if (::bind(fd, saddr, address.getActualSize()) != 0) { if (::bind(fd, saddr, address.getActualSize()) != 0) {
if (!isExistingSocket) { if (!isExistingSocket) {
::close(fd); closeNoInt(fd);
} }
folly::throwSystemError(errno, folly::throwSystemError(errno,
"failed to bind to async server socket: " + "failed to bind to async server socket: " +
...@@ -360,9 +361,7 @@ void AsyncServerSocket::bind(uint16_t port) { ...@@ -360,9 +361,7 @@ void AsyncServerSocket::bind(uint16_t port) {
"bad getaddrinfo"); "bad getaddrinfo");
} }
folly::ScopeGuard guard = folly::makeGuard([&]{ SCOPE_EXIT { freeaddrinfo(res0); };
freeaddrinfo(res0);
});
auto setupAddress = [&] (struct addrinfo* res) { auto setupAddress = [&] (struct addrinfo* res) {
int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol); int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
...@@ -375,7 +374,7 @@ void AsyncServerSocket::bind(uint16_t port) { ...@@ -375,7 +374,7 @@ void AsyncServerSocket::bind(uint16_t port) {
try { try {
setupSocket(s); setupSocket(s);
} catch (...) { } catch (...) {
::close(s); closeNoInt(s);
throw; throw;
} }
...@@ -399,6 +398,7 @@ void AsyncServerSocket::bind(uint16_t port) { ...@@ -399,6 +398,7 @@ void AsyncServerSocket::bind(uint16_t port) {
} }
}; };
const int kNumTries = 5;
for (int tries = 1; true; tries++) { for (int tries = 1; true; tries++) {
// Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
// should return IPv6 first and then IPv4 addresses, but glibc's // should return IPv6 first and then IPv4 addresses, but glibc's
...@@ -413,8 +413,9 @@ void AsyncServerSocket::bind(uint16_t port) { ...@@ -413,8 +413,9 @@ void AsyncServerSocket::bind(uint16_t port) {
} }
// If port == 0, then we should try to bind to the same port on ipv4 and // 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. // ipv6. So if we did bind to ipv6, figure out that port and use it,
if (!sockets_.empty() && port == 0) { // except for the last attempt when we just use any port available.
if (sockets_.size() == 1 && port == 0) {
SocketAddress address; SocketAddress address;
address.setFromLocalAddress(sockets_.back().socket_); address.setFromLocalAddress(sockets_.back().socket_);
snprintf(sport, sizeof(sport), "%u", address.getPort()); snprintf(sport, sizeof(sport), "%u", address.getPort());
...@@ -425,17 +426,21 @@ void AsyncServerSocket::bind(uint16_t port) { ...@@ -425,17 +426,21 @@ void AsyncServerSocket::bind(uint16_t port) {
try { try {
for (res = res0; res; res = res->ai_next) { for (res = res0; res; res = res->ai_next) {
if (res->ai_family != AF_INET6) { if (res->ai_family != AF_INET6) {
setupAddress(res); setupAddress(res);
} }
} }
} catch (const std::system_error& e) { } catch (const std::system_error& e) {
// if we can't bind to the same port on ipv4 as ipv6 when using port=0 // if we can't bind to the same port on ipv4 as ipv6 when using port=0
// then we will try again another 2 times before giving up. We do this // then we will try again another 2 times before giving up. We do this
// by closing the sockets that were opened, then redoing the whole thing // by closing the sockets that were opened, then redoing the whole thing
if (port == 0 && !sockets_.empty() && tries != 3) { if (port == 0 && !sockets_.empty() && tries != kNumTries) {
for (const auto& socket : sockets_) { for (const auto& socket : sockets_) {
if (socket.socket_ > 0) { if (socket.socket_ <= 0) {
CHECK_EQ(0, ::close(socket.socket_)); continue;
} else if (shutdownSocketSet_) {
shutdownSocketSet_->close(socket.socket_);
} else {
closeNoInt(socket.socket_);
} }
} }
sockets_.clear(); sockets_.clear();
...@@ -625,7 +630,7 @@ int AsyncServerSocket::createSocket(int family) { ...@@ -625,7 +630,7 @@ int AsyncServerSocket::createSocket(int family) {
try { try {
setupSocket(fd); setupSocket(fd);
} catch (...) { } catch (...) {
::close(fd); closeNoInt(fd);
throw; throw;
} }
return fd; return fd;
...@@ -735,7 +740,7 @@ void AsyncServerSocket::handlerReady( ...@@ -735,7 +740,7 @@ void AsyncServerSocket::handlerReady(
} else if (rand() > acceptRate_ * RAND_MAX) { } else if (rand() > acceptRate_ * RAND_MAX) {
++numDroppedConnections_; ++numDroppedConnections_;
if (clientSocket >= 0) { if (clientSocket >= 0) {
::close(clientSocket); closeNoInt(clientSocket);
} }
continue; continue;
} }
...@@ -765,7 +770,7 @@ void AsyncServerSocket::handlerReady( ...@@ -765,7 +770,7 @@ void AsyncServerSocket::handlerReady(
#ifndef SOCK_NONBLOCK #ifndef SOCK_NONBLOCK
// Explicitly set the new connection to non-blocking mode // Explicitly set the new connection to non-blocking mode
if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) { if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
::close(clientSocket); closeNoInt(clientSocket);
dispatchError("failed to set accepted socket to non-blocking mode", dispatchError("failed to set accepted socket to non-blocking mode",
errno); errno);
return; return;
...@@ -829,7 +834,7 @@ void AsyncServerSocket::dispatchSocket(int socket, ...@@ -829,7 +834,7 @@ void AsyncServerSocket::dispatchSocket(int socket,
// even accept new messages. // even accept new messages.
LOG(ERROR) << "failed to dispatch newly accepted socket:" LOG(ERROR) << "failed to dispatch newly accepted socket:"
<< " all accept callback queues are full"; << " all accept callback queues are full";
::close(socket); closeNoInt(socket);
return; return;
} }
......
...@@ -222,6 +222,7 @@ AsyncSocket::AsyncSocket(EventBase* evb, int fd) ...@@ -222,6 +222,7 @@ AsyncSocket::AsyncSocket(EventBase* evb, int fd)
<< fd << ")"; << fd << ")";
init(); init();
fd_ = fd; fd_ = fd;
setCloseOnExec();
state_ = StateEnum::ESTABLISHED; state_ = StateEnum::ESTABLISHED;
} }
...@@ -293,6 +294,15 @@ void AsyncSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) { ...@@ -293,6 +294,15 @@ void AsyncSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
} }
} }
void AsyncSocket::setCloseOnExec() {
int rv = fcntl(fd_, F_SETFD, FD_CLOEXEC);
if (rv != 0) {
throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
withAddr("failed to set close-on-exec flag"),
errno);
}
}
void AsyncSocket::connect(ConnectCallback* callback, void AsyncSocket::connect(ConnectCallback* callback,
const folly::SocketAddress& address, const folly::SocketAddress& address,
int timeout, int timeout,
...@@ -331,14 +341,7 @@ void AsyncSocket::connect(ConnectCallback* callback, ...@@ -331,14 +341,7 @@ void AsyncSocket::connect(ConnectCallback* callback,
} }
ioHandler_.changeHandlerFD(fd_); ioHandler_.changeHandlerFD(fd_);
// Set the FD_CLOEXEC flag so that the socket will be closed if the program setCloseOnExec();
// later forks and execs.
int rv = fcntl(fd_, F_SETFD, FD_CLOEXEC);
if (rv != 0) {
throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
withAddr("failed to set close-on-exec flag"),
errno);
}
// Put the socket in non-blocking mode // Put the socket in non-blocking mode
int flags = fcntl(fd_, F_GETFL, 0); int flags = fcntl(fd_, F_GETFL, 0);
...@@ -346,7 +349,7 @@ void AsyncSocket::connect(ConnectCallback* callback, ...@@ -346,7 +349,7 @@ void AsyncSocket::connect(ConnectCallback* callback,
throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR, throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
withAddr("failed to get socket flags"), errno); withAddr("failed to get socket flags"), errno);
} }
rv = fcntl(fd_, F_SETFL, flags | O_NONBLOCK); int rv = fcntl(fd_, F_SETFL, flags | O_NONBLOCK);
if (rv == -1) { if (rv == -1) {
throw AsyncSocketException( throw AsyncSocketException(
AsyncSocketException::INTERNAL_ERROR, AsyncSocketException::INTERNAL_ERROR,
......
...@@ -398,6 +398,13 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -398,6 +398,13 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
*/ */
int setNoDelay(bool noDelay); int setNoDelay(bool noDelay);
/**
* Set the FD_CLOEXEC flag so that the socket will be closed if the program
* later forks and execs.
*/
void setCloseOnExec();
/* /*
* Set the Flavor of Congestion Control to be used for this Socket * Set the Flavor of Congestion Control to be used for this Socket
* Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <folly/io/async/EventBase.h> #include <folly/io/async/EventBase.h>
#include <folly/Baton.h>
#include <folly/ThreadName.h> #include <folly/ThreadName.h>
#include <folly/io/async/NotificationQueue.h> #include <folly/io/async/NotificationQueue.h>
...@@ -562,6 +563,40 @@ bool EventBase::runInEventBaseThread(const Cob& fn) { ...@@ -562,6 +563,40 @@ bool EventBase::runInEventBaseThread(const Cob& fn) {
return true; return true;
} }
bool EventBase::runInEventBaseThreadAndWait(void (*fn)(void*), void* arg) {
if (inRunningEventBaseThread()) {
LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
<< "allowed";
return false;
}
Baton<> ready;
runInEventBaseThread([&] {
fn(arg);
ready.post();
});
ready.wait();
return true;
}
bool EventBase::runInEventBaseThreadAndWait(const Cob& fn) {
if (inRunningEventBaseThread()) {
LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
<< "allowed";
return false;
}
Baton<> ready;
runInEventBaseThread([&] {
fn();
ready.post();
});
ready.wait();
return true;
}
bool EventBase::runAfterDelay(const Cob& cob, bool EventBase::runAfterDelay(const Cob& cob,
int milliseconds, int milliseconds,
TimeoutManager::InternalEnum in) { TimeoutManager::InternalEnum in) {
......
...@@ -346,6 +346,28 @@ class EventBase : private boost::noncopyable, ...@@ -346,6 +346,28 @@ class EventBase : private boost::noncopyable,
*/ */
bool runInEventBaseThread(const Cob& fn); bool runInEventBaseThread(const Cob& fn);
/*
* Like runInEventBaseThread, but the caller waits for the callback to be
* executed.
*/
template<typename T>
bool runInEventBaseThreadAndWait(void (*fn)(T*), T* arg) {
return runInEventBaseThreadAndWait(reinterpret_cast<void (*)(void*)>(fn),
reinterpret_cast<void*>(arg));
}
/*
* Like runInEventBaseThread, but the caller waits for the callback to be
* executed.
*/
bool runInEventBaseThreadAndWait(void (*fn)(void*), void* arg);
/*
* Like runInEventBaseThread, but the caller waits for the callback to be
* executed.
*/
bool runInEventBaseThreadAndWait(const Cob& fn);
/** /**
* Runs the given Cob at some time after the specified number of * Runs the given Cob at some time after the specified number of
* milliseconds. (No guarantees exactly when.) * milliseconds. (No guarantees exactly when.)
......
...@@ -173,10 +173,8 @@ a lock on accept()ing from a port, preventing more than ~20k accepts / ...@@ -173,10 +173,8 @@ a lock on accept()ing from a port, preventing more than ~20k accepts /
sec. There are various workarounds (SO_REUSEPORT), but generally sec. There are various workarounds (SO_REUSEPORT), but generally
clients should be using connection pooling instead when possible. clients should be using connection pooling instead when possible.
#### AsyncSSLServerSocket Since AsyncServerSocket provides an fd, an AsyncSSLSocket or
AsyncSocket can be made using the same codepath
Similar to AsyncServerSocket, but provides callbacks for SSL
handshaking.
#### TAsyncUDPServerSocket #### TAsyncUDPServerSocket
......
...@@ -20,17 +20,8 @@ ...@@ -20,17 +20,8 @@
*/ */
#include <folly/io/async/Request.h> #include <folly/io/async/Request.h>
#ifndef NO_LIB_GFLAGS
DEFINE_bool(enable_request_context, true,
"Enable collection of per-request queueing stats for thrift");
#endif
namespace folly { namespace folly {
#ifdef NO_LIB_GFLAGS
bool FLAGS_enable_request_context = true;
#endif
RequestContext* defaultContext; RequestContext* defaultContext;
} }
...@@ -26,25 +26,8 @@ ...@@ -26,25 +26,8 @@
#include <folly/ThreadLocal.h> #include <folly/ThreadLocal.h>
#include <folly/RWSpinLock.h> #include <folly/RWSpinLock.h>
/**
* In many cases this header is included as a
* dependency to libraries which do not need
* command line flags. GFLAGS is a large binary
* and thus we do this so that a library which
* is size sensitive doesn't have to pull in
* GFLAGS if it doesn't want to.
*/
#ifndef NO_LIB_GFLAGS
#include <gflags/gflags.h>
DECLARE_bool(enable_request_context);
#endif
namespace folly { namespace folly {
#ifdef NO_LIB_GFLAGS
extern bool FLAGS_enable_request_context;
#endif
// Some request context that follows an async request through a process // Some request context that follows an async request through a process
// Everything in the context must be thread safe // Everything in the context must be thread safe
...@@ -66,9 +49,6 @@ class RequestContext { ...@@ -66,9 +49,6 @@ class RequestContext {
// It will be passed between queues / threads (where implemented), // It will be passed between queues / threads (where implemented),
// so it should be valid for the lifetime of the request. // so it should be valid for the lifetime of the request.
static void create() { static void create() {
if(!FLAGS_enable_request_context) {
return;
}
getStaticContext() = std::make_shared<RequestContext>(); getStaticContext() = std::make_shared<RequestContext>();
} }
...@@ -88,10 +68,6 @@ class RequestContext { ...@@ -88,10 +68,6 @@ class RequestContext {
// profiling any use of these functions. // profiling any use of these functions.
void setContextData( void setContextData(
const std::string& val, std::unique_ptr<RequestData> data) { const std::string& val, std::unique_ptr<RequestData> data) {
if (!FLAGS_enable_request_context) {
return;
}
folly::RWSpinLock::WriteHolder guard(lock); folly::RWSpinLock::WriteHolder guard(lock);
if (data_.find(val) != data_.end()) { if (data_.find(val) != data_.end()) {
LOG_FIRST_N(WARNING, 1) << LOG_FIRST_N(WARNING, 1) <<
...@@ -132,15 +108,12 @@ class RequestContext { ...@@ -132,15 +108,12 @@ class RequestContext {
static std::shared_ptr<RequestContext> static std::shared_ptr<RequestContext>
setContext(std::shared_ptr<RequestContext> ctx) { setContext(std::shared_ptr<RequestContext> ctx) {
if (FLAGS_enable_request_context) { std::shared_ptr<RequestContext> old_ctx;
std::shared_ptr<RequestContext> old_ctx; if (getStaticContext()) {
if (getStaticContext()) { old_ctx = getStaticContext();
old_ctx = getStaticContext();
}
getStaticContext() = ctx;
return old_ctx;
} }
return nullptr; getStaticContext() = ctx;
return old_ctx;
} }
static std::shared_ptr<RequestContext> saveContext() { static std::shared_ptr<RequestContext> saveContext() {
......
...@@ -22,14 +22,17 @@ ...@@ -22,14 +22,17 @@
#include <folly/io/async/test/SocketPair.h> #include <folly/io/async/test/SocketPair.h>
#include <folly/io/async/test/Util.h> #include <folly/io/async/test/Util.h>
#include <atomic>
#include <iostream> #include <iostream>
#include <unistd.h> #include <unistd.h>
#include <memory> #include <memory>
#include <thread> #include <thread>
using std::atomic;
using std::deque; using std::deque;
using std::pair; using std::pair;
using std::vector; using std::vector;
using std::thread;
using std::make_pair; using std::make_pair;
using std::cerr; using std::cerr;
using std::endl; using std::endl;
...@@ -1171,6 +1174,45 @@ TEST(EventBaseTest, RunInThread) { ...@@ -1171,6 +1174,45 @@ TEST(EventBaseTest, RunInThread) {
} }
} }
// This test simulates some calls, and verifies that the waiting happens by
// triggering what otherwise would be race conditions, and trying to detect
// whether any of the race conditions happened.
TEST(EventBaseTest, RunInEventLoopThreadAndWait) {
const size_t c = 256;
vector<atomic<size_t>> atoms(c);
for (size_t i = 0; i < c; ++i) {
auto& atom = atoms.at(i);
atom = 0;
}
vector<thread> threads(c);
for (size_t i = 0; i < c; ++i) {
auto& atom = atoms.at(i);
auto& th = threads.at(i);
th = thread([&atom] {
EventBase eb;
auto ebth = thread([&]{ eb.loopForever(); });
eb.waitUntilRunning();
eb.runInEventBaseThreadAndWait([&] {
size_t x = 0;
atom.compare_exchange_weak(
x, 1, std::memory_order_release, std::memory_order_relaxed);
});
size_t x = 0;
atom.compare_exchange_weak(
x, 2, std::memory_order_release, std::memory_order_relaxed);
eb.terminateLoopSoon();
ebth.join();
});
}
for (size_t i = 0; i < c; ++i) {
auto& th = threads.at(i);
th.join();
}
size_t sum = 0;
for (auto& atom : atoms) sum += atom;
EXPECT_EQ(c, sum);
}
/////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////
// Tests for runInLoop() // Tests for runInLoop()
/////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment