From: Dave Watson <davejwatson@fb.com>
Date: Thu, 5 Mar 2015 18:25:54 +0000 (-0800)
Subject: httpserver on serverbootstrap
X-Git-Tag: v0.29.0~1
X-Git-Url: http://demsky.eecs.uci.edu/git/?a=commitdiff_plain;h=321807e1014b6b5cf8bbf68e21cb8fb996ab361b;p=folly.git

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
---

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