Commit 321807e1 authored by Dave Watson's avatar Dave Watson Committed by Nicholas Ormrod

httpserver on serverbootstrap

Summary:
Cleans up the httpserver startup code nicely.  The only major change to ServerBootstrap was a check if bind failed to throw an exception.

(depends on D1732895)

Test Plan:
fbconfig -r folly/wangle/bootstrap proxygen/httpserver; fbmake runtests

fbconfig -r dfsrouter; fbmake runtests

Reviewed By: hans@fb.com

Subscribers: yfeldblum, cgheorghe, trunkagent, doug, fugalh, bmatheny, folly-diffs@, jsedgwick

FB internal diff: D1800100

Signature: t1:1800100:1424733970:67a61a22d2affadea16d2fd725003915326077b2
parent b252ef02
...@@ -92,7 +92,7 @@ class Acceptor : ...@@ -92,7 +92,7 @@ class Acceptor :
*/ */
uint32_t getNumConnections() const { uint32_t getNumConnections() const {
return downstreamConnectionManager_ ? return downstreamConnectionManager_ ?
downstreamConnectionManager_->getNumConnections() : 0; (uint32_t)downstreamConnectionManager_->getNumConnections() : 0;
} }
/** /**
......
...@@ -20,6 +20,9 @@ ...@@ -20,6 +20,9 @@
namespace folly { namespace folly {
typedef folly::wangle::ChannelPipeline<
folly::IOBufQueue&, std::unique_ptr<folly::IOBuf>> DefaultPipeline;
/* /*
* ServerBootstrap is a parent class intended to set up a * ServerBootstrap is a parent class intended to set up a
* high-performance TCP accepting server. It will manage a pool of * high-performance TCP accepting server. It will manage a pool of
...@@ -60,8 +63,8 @@ class ServerBootstrap { ...@@ -60,8 +63,8 @@ class ServerBootstrap {
* *
* @param childHandler - acceptor factory to call for each IO thread * @param childHandler - acceptor factory to call for each IO thread
*/ */
ServerBootstrap* childHandler(std::shared_ptr<AcceptorFactory> childHandler) { ServerBootstrap* childHandler(std::shared_ptr<AcceptorFactory> h) {
acceptorFactory_ = childHandler; acceptorFactory_ = h;
return this; return this;
} }
...@@ -162,7 +165,7 @@ class ServerBootstrap { ...@@ -162,7 +165,7 @@ class ServerBootstrap {
sockets_.push_back(socket); sockets_.push_back(socket);
} }
void bind(folly::SocketAddress address) { void bind(folly::SocketAddress& address) {
bindImpl(-1, address); bindImpl(-1, address);
} }
...@@ -174,10 +177,11 @@ class ServerBootstrap { ...@@ -174,10 +177,11 @@ class ServerBootstrap {
*/ */
void bind(int port) { void bind(int port) {
CHECK(port >= 0); CHECK(port >= 0);
bindImpl(port, folly::SocketAddress()); folly::SocketAddress address;
bindImpl(port, address);
} }
void bindImpl(int port, folly::SocketAddress address) { void bindImpl(int port, folly::SocketAddress& address) {
if (!workerFactory_) { if (!workerFactory_) {
group(nullptr); group(nullptr);
} }
...@@ -190,24 +194,35 @@ class ServerBootstrap { ...@@ -190,24 +194,35 @@ class ServerBootstrap {
std::mutex sock_lock; std::mutex sock_lock;
std::vector<std::shared_ptr<folly::AsyncServerSocket>> new_sockets; std::vector<std::shared_ptr<folly::AsyncServerSocket>> new_sockets;
std::exception_ptr exn;
auto startupFunc = [&](std::shared_ptr<folly::Baton<>> barrier){ auto startupFunc = [&](std::shared_ptr<folly::Baton<>> barrier){
auto socket = folly::AsyncServerSocket::newSocket(); auto socket = folly::AsyncServerSocket::newSocket();
sock_lock.lock();
new_sockets.push_back(socket);
sock_lock.unlock();
socket->setReusePortEnabled(reusePort); socket->setReusePortEnabled(reusePort);
socket->attachEventBase(EventBaseManager::get()->getEventBase()); socket->attachEventBase(EventBaseManager::get()->getEventBase());
if (port >= 0) {
socket->bind(port); try {
} else { if (port >= 0) {
socket->bind(address); socket->bind(port);
port = address.getPort(); } else {
socket->bind(address);
port = address.getPort();
}
socket->listen(socketConfig.acceptBacklog);
socket->startAccepting();
} catch (...) {
exn = std::current_exception();
barrier->post();
return;
} }
socket->listen(socketConfig.acceptBacklog);
socket->startAccepting(); sock_lock.lock();
new_sockets.push_back(socket);
sock_lock.unlock();
if (port == 0) { if (port == 0) {
SocketAddress address;
socket->getAddress(&address); socket->getAddress(&address);
port = address.getPort(); port = address.getPort();
} }
...@@ -225,6 +240,10 @@ class ServerBootstrap { ...@@ -225,6 +240,10 @@ class ServerBootstrap {
barrier->wait(); barrier->wait();
} }
if (exn) {
std::rethrow_exception(exn);
}
// Startup all the threads // Startup all the threads
for(auto socket : new_sockets) { for(auto socket : new_sockets) {
workerFactory_->forEachWorker([this, socket](Acceptor* worker){ workerFactory_->forEachWorker([this, socket](Acceptor* worker){
......
...@@ -282,8 +282,8 @@ class ChannelPipeline<R, W, Handler, Handlers...> ...@@ -282,8 +282,8 @@ class ChannelPipeline<R, W, Handler, Handlers...>
ChannelPipeline<R, W, Handlers...>::finalizeHelper(); ChannelPipeline<R, W, Handlers...>::finalizeHelper();
back_ = ChannelPipeline<R, W, Handlers...>::back_; back_ = ChannelPipeline<R, W, Handlers...>::back_;
if (!back_) { if (!back_) {
auto is_end = ChannelPipeline<R, W, Handlers...>::is_end; auto is_at_end = ChannelPipeline<R, W, Handlers...>::is_end;
CHECK(is_end); CHECK(is_at_end);
back_ = dynamic_cast<OutboundChannelHandlerContext<W>*>(&ctx_); back_ = dynamic_cast<OutboundChannelHandlerContext<W>*>(&ctx_);
if (!back_) { if (!back_) {
throw std::invalid_argument("wrong type for last handler"); throw std::invalid_argument("wrong type for last handler");
......
...@@ -41,7 +41,7 @@ class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor { ...@@ -41,7 +41,7 @@ class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor {
EventBase* getEventBase() override; EventBase* getEventBase() override;
EventBase* getEventBase(ThreadPoolExecutor::ThreadHandle*); static EventBase* getEventBase(ThreadPoolExecutor::ThreadHandle*);
private: private:
struct FOLLY_ALIGN_TO_AVOID_FALSE_SHARING IOThread : public Thread { struct FOLLY_ALIGN_TO_AVOID_FALSE_SHARING IOThread : public Thread {
......
...@@ -104,8 +104,12 @@ class ThreadPoolExecutor : public virtual Executor { ...@@ -104,8 +104,12 @@ class ThreadPoolExecutor : public virtual Executor {
public: public:
virtual void threadStarted(ThreadHandle*) = 0; virtual void threadStarted(ThreadHandle*) = 0;
virtual void threadStopped(ThreadHandle*) = 0; virtual void threadStopped(ThreadHandle*) = 0;
virtual void threadPreviouslyStarted(ThreadHandle*) = 0; virtual void threadPreviouslyStarted(ThreadHandle* h) {
virtual void threadNotYetStopped(ThreadHandle*) = 0; threadStarted(h);
}
virtual void threadNotYetStopped(ThreadHandle* h) {
threadStopped(h);
}
virtual ~Observer() = default; virtual ~Observer() = default;
}; };
......
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