wangle::ServerBootstrap< Pipeline > Class Template Reference

#include <ServerBootstrap.h>

Public Member Functions
	ServerBootstrap (const ServerBootstrap &that)=delete
	ServerBootstrap (ServerBootstrap &&that)=default
	ServerBootstrap ()=default
	~ServerBootstrap ()
ServerBootstrap *	pipeline (std::shared_ptr< AcceptPipelineFactory > factory)
ServerBootstrap *	channelFactory (std::shared_ptr< ServerSocketFactory > factory)
ServerBootstrap *	acceptorConfig (const ServerSocketConfig &accConfig)
ServerBootstrap *	childHandler (std::shared_ptr< AcceptorFactory > h)
ServerBootstrap *	childPipeline (std::shared_ptr< PipelineFactory< Pipeline >> factory)
ServerBootstrap *	group (std::shared_ptr< folly::IOThreadPoolExecutor > io_group)
ServerBootstrap *	group (std::shared_ptr< folly::IOThreadPoolExecutor > accept_group, std::shared_ptr< folly::IOThreadPoolExecutor > io_group)
void	bind (folly::AsyncServerSocket::UniquePtr s)
void	bind (folly::SocketAddress &address)
void	bind (int port)
void	bindImpl (folly::SocketAddress &address)
void	stop ()
void	join ()
void	waitForStop ()
const std::vector< std::shared_ptr< folly::AsyncSocketBase > > &	getSockets () const
std::shared_ptr< folly::IOThreadPoolExecutor >	getIOGroup () const
template<typename F >
void	forEachWorker (F &&f) const
ServerBootstrap *	setReusePort (bool reusePort)

Public Attributes
ServerSocketConfig	socketConfig

Private Attributes
std::shared_ptr< folly::IOThreadPoolExecutor >	acceptor_group_
std::shared_ptr< folly::IOThreadPoolExecutor >	io_group_
std::shared_ptr< ServerWorkerPool >	workerFactory_
std::shared_ptr< std::vector< std::shared_ptr< folly::AsyncSocketBase > > >	sockets_
std::shared_ptr< AcceptorFactory >	acceptorFactory_
std::shared_ptr< PipelineFactory< Pipeline > >	childPipelineFactory_
std::shared_ptr< AcceptPipelineFactory >	acceptPipelineFactory_
std::shared_ptr< ServerSocketFactory >	socketFactory_
ServerSocketConfig	accConfig_
bool	reusePort_ {false}
std::unique_ptr< folly::Baton<> >	stopBaton_
bool	stopped_ {false}

Detailed Description

template<typename Pipeline = wangle::DefaultPipeline>
class wangle::ServerBootstrap< Pipeline >

Definition at line 42 of file ServerBootstrap.h.

Constructor & Destructor Documentation

template<typename Pipeline = wangle::DefaultPipeline>

wangle::ServerBootstrap< Pipeline >::ServerBootstrap ( const ServerBootstrap< Pipeline > &  that )

delete

template<typename Pipeline = wangle::DefaultPipeline>

wangle::ServerBootstrap< Pipeline >::ServerBootstrap ( ServerBootstrap< Pipeline > &&  that )

default

template<typename Pipeline = wangle::DefaultPipeline>

wangle::ServerBootstrap< Pipeline >::ServerBootstrap ( )

default

template<typename Pipeline = wangle::DefaultPipeline>

wangle::ServerBootstrap< Pipeline >::~ServerBootstrap ( )

inline

Definition at line 49 of file ServerBootstrap.h.

                     {
    stop();
    join();
  }

Member Function Documentation

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::acceptorConfig ( const ServerSocketConfig &  accConfig )

inline

Definition at line 70 of file ServerBootstrap.h.

Referenced by main().

                                                                       {
    accConfig_ = accConfig;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::bind ( folly::AsyncServerSocket::UniquePtr  s )

inline

Definition at line 163 of file ServerBootstrap.h.

Referenced by main(), and wangle::TEST().

                                               {
    if (!workerFactory_) {
      group(nullptr);
    }

    // Since only a single socket is given,
    // we can only accept on a single thread
    CHECK(acceptor_group_->numThreads() == 1);

    std::shared_ptr<folly::AsyncServerSocket> socket(
      s.release(), AsyncServerSocketFactory::ThreadSafeDestructor());
    socket->setMaxNumMessagesInQueue(
        accConfig_.maxNumPendingConnectionsPerWorker);

    folly::via(acceptor_group_.get(), [&] {
      socket->attachEventBase(folly::EventBaseManager::get()->getEventBase());
      socket->listen(socketConfig.acceptBacklog);
      socket->startAccepting();
    }).get();

    // Startup all the threads
    workerFactory_->forEachWorker([this, socket](Acceptor* worker){
      socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
        [this, worker, socket](){
          socketFactory_->addAcceptCB(socket, worker, worker->getEventBase());
      });
    });

    sockets_->push_back(socket);
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::bind ( folly::SocketAddress &  address )

inline

Definition at line 194 of file ServerBootstrap.h.

                                         {
    bindImpl(address);
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::bind ( int  port )

inline

Definition at line 204 of file ServerBootstrap.h.

                      {
    CHECK(port >= 0);
    folly::SocketAddress address;
    address.setFromLocalPort(port);
    bindImpl(address);
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::bindImpl ( folly::SocketAddress &  address )

inline

Definition at line 211 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind().

                                             {
    if (!workerFactory_) {
      group(nullptr);
    }

    bool reusePort = reusePort_ || (acceptor_group_->numThreads() > 1);

    std::mutex sock_lock;
    std::vector<std::shared_ptr<folly::AsyncSocketBase>> new_sockets;


    std::exception_ptr exn;

    auto startupFunc = [&](std::shared_ptr<folly::Baton<>> barrier) {

      try {
        auto socket = socketFactory_->newSocket(
            address, socketConfig.acceptBacklog, reusePort, socketConfig);
        sock_lock.lock();
        new_sockets.push_back(socket);
        sock_lock.unlock();
        socket->getAddress(&address);

        barrier->post();
      } catch (...) {
        exn = std::current_exception();
        barrier->post();

        return;
      }

    };

    auto wait0 = std::make_shared<folly::Baton<>>();
    acceptor_group_->add(std::bind(startupFunc, wait0));
    wait0->wait();

    for (size_t i = 1; i < acceptor_group_->numThreads(); i++) {
      auto barrier = std::make_shared<folly::Baton<>>();
      acceptor_group_->add(std::bind(startupFunc, barrier));
      barrier->wait();
    }

    if (exn) {
      std::rethrow_exception(exn);
    }

    for (auto& socket : new_sockets) {
      // Startup all the threads
      workerFactory_->forEachWorker([this, socket](Acceptor* worker){
        socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
          [this, worker, socket](){
            socketFactory_->addAcceptCB(socket, worker, worker->getEventBase());
        });
      });

      sockets_->push_back(socket);
    }
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::channelFactory ( std::shared_ptr< ServerSocketFactory >  factory )

inline

Definition at line 64 of file ServerBootstrap.h.

                                                {
    socketFactory_ = factory;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::childHandler ( std::shared_ptr< AcceptorFactory >  h )

inline

Definition at line 81 of file ServerBootstrap.h.

                                                                  {
    acceptorFactory_ = h;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::childPipeline ( std::shared_ptr< PipelineFactory< Pipeline >>  factory )

inline

Definition at line 91 of file ServerBootstrap.h.

Referenced by main(), and wangle::TEST().

                                                        {
    childPipelineFactory_ = factory;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

template<typename F >

void wangle::ServerBootstrap< Pipeline >::forEachWorker ( F &&  f ) const

inline

Definition at line 317 of file ServerBootstrap.h.

                                  {
    if (!workerFactory_) {
      return;
    }
    workerFactory_->forEachWorker(f);
  }

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<folly::IOThreadPoolExecutor> wangle::ServerBootstrap< Pipeline >::getIOGroup ( ) const

inline

Definition at line 312 of file ServerBootstrap.h.

                                                              {
    return io_group_;
  }

template<typename Pipeline = wangle::DefaultPipeline>

const std::vector<std::shared_ptr<folly::AsyncSocketBase> >& wangle::ServerBootstrap< Pipeline >::getSockets ( ) const

inline

Definition at line 308 of file ServerBootstrap.h.

Referenced by wangle::TEST().

                     {
    return *sockets_;
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::group ( std::shared_ptr< folly::IOThreadPoolExecutor >  io_group )

inline

Definition at line 103 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), wangle::ServerBootstrap< DefaultPipeline >::group(), and main().

                                                         {
    return group(nullptr, io_group);
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::group ( std::shared_ptr< folly::IOThreadPoolExecutor >  accept_group, std::shared_ptr< folly::IOThreadPoolExecutor >  io_group  )

inline

Definition at line 117 of file ServerBootstrap.h.

                                                         {
    if (!accept_group) {
      accept_group = std::make_shared<folly::IOThreadPoolExecutor>(
        1, std::make_shared<folly::NamedThreadFactory>("Acceptor Thread"));
    }
    if (!io_group) {
      auto threads = std::thread::hardware_concurrency();
      if (threads <= 0) {
        // Reasonable mid-point for concurrency when actual value unknown
        threads = 8;
      }
      io_group = std::make_shared<folly::IOThreadPoolExecutor>(
        threads, std::make_shared<folly::NamedThreadFactory>("IO Thread"));
    }

    // TODO better config checking
    // CHECK(acceptorFactory_ || childPipelineFactory_);
    CHECK(!(acceptorFactory_ && childPipelineFactory_));

    if (acceptorFactory_) {
      workerFactory_ = std::make_shared<ServerWorkerPool>(
        acceptorFactory_, io_group.get(), sockets_, socketFactory_);
    } else {
      workerFactory_ = std::make_shared<ServerWorkerPool>(
          std::make_shared<ServerAcceptorFactory<Pipeline>>(
              acceptPipelineFactory_, childPipelineFactory_, accConfig_),
          io_group.get(),
          sockets_,
          socketFactory_);
    }

    io_group->addObserver(workerFactory_);

    acceptor_group_ = accept_group;
    io_group_ = io_group;

    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::join ( )

inline

Definition at line 288 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::~ServerBootstrap().

              {
    if (acceptor_group_) {
      acceptor_group_->join();
    }
    if (io_group_) {
      io_group_->join();
    }
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::pipeline ( std::shared_ptr< AcceptPipelineFactory >  factory )

inline

Definition at line 59 of file ServerBootstrap.h.

Referenced by main().

                                                                          {
    acceptPipelineFactory_ = factory;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

ServerBootstrap* wangle::ServerBootstrap< Pipeline >::setReusePort ( bool  reusePort )

inline

Definition at line 326 of file ServerBootstrap.h.

                                                {
    reusePort_ = reusePort;
    return this;
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::stop ( )

inline

Definition at line 274 of file ServerBootstrap.h.

Referenced by wangle::TEST(), and wangle::ServerBootstrap< DefaultPipeline >::~ServerBootstrap().

              {
    // sockets_ may be null if ServerBootstrap has been std::move'd
    if (sockets_) {
      sockets_->clear();
    }
    if (!stopped_) {
      stopped_ = true;
      // stopBaton_ may be null if ServerBootstrap has been std::move'd
      if (stopBaton_) {
        stopBaton_->post();
      }
    }
  }

template<typename Pipeline = wangle::DefaultPipeline>

void wangle::ServerBootstrap< Pipeline >::waitForStop ( )

inline

Definition at line 297 of file ServerBootstrap.h.

Referenced by main().

                     {
    if (!stopped_) {
      CHECK(stopBaton_);
      stopBaton_->wait();
    }
  }

Member Data Documentation

template<typename Pipeline = wangle::DefaultPipeline>

ServerSocketConfig wangle::ServerBootstrap< Pipeline >::accConfig_

private

Definition at line 346 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::acceptorConfig(), wangle::ServerBootstrap< DefaultPipeline >::bind(), and wangle::ServerBootstrap< DefaultPipeline >::group().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<folly::IOThreadPoolExecutor> wangle::ServerBootstrap< Pipeline >::acceptor_group_

private

Definition at line 332 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), wangle::ServerBootstrap< DefaultPipeline >::group(), and wangle::ServerBootstrap< DefaultPipeline >::join().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<AcceptorFactory> wangle::ServerBootstrap< Pipeline >::acceptorFactory_

private

Definition at line 339 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::childHandler(), and wangle::ServerBootstrap< DefaultPipeline >::group().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<AcceptPipelineFactory> wangle::ServerBootstrap< Pipeline >::acceptPipelineFactory_

private

Initial value:

{
      std::make_shared<DefaultAcceptPipelineFactory>()}

Definition at line 341 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::group(), and wangle::ServerBootstrap< DefaultPipeline >::pipeline().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<PipelineFactory<Pipeline> > wangle::ServerBootstrap< Pipeline >::childPipelineFactory_

private

Definition at line 340 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::childPipeline(), and wangle::ServerBootstrap< DefaultPipeline >::group().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<folly::IOThreadPoolExecutor> wangle::ServerBootstrap< Pipeline >::io_group_

private

Definition at line 333 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::getIOGroup(), wangle::ServerBootstrap< DefaultPipeline >::group(), and wangle::ServerBootstrap< DefaultPipeline >::join().

template<typename Pipeline = wangle::DefaultPipeline>

bool wangle::ServerBootstrap< Pipeline >::reusePort_ {false}

private

Definition at line 348 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), and wangle::ServerBootstrap< DefaultPipeline >::setReusePort().

template<typename Pipeline = wangle::DefaultPipeline>

ServerSocketConfig wangle::ServerBootstrap< Pipeline >::socketConfig

Definition at line 324 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), and wangle::ServerBootstrap< DefaultPipeline >::bindImpl().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<ServerSocketFactory> wangle::ServerBootstrap< Pipeline >::socketFactory_

private

Initial value:

{
    std::make_shared<AsyncServerSocketFactory>()}

Definition at line 343 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), wangle::ServerBootstrap< DefaultPipeline >::channelFactory(), and wangle::ServerBootstrap< DefaultPipeline >::group().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<std::vector<std::shared_ptr<folly::AsyncSocketBase> > > wangle::ServerBootstrap< Pipeline >::sockets_

private

Initial value:

{
    std::make_shared<std::vector<std::shared_ptr<folly::AsyncSocketBase>>>()}

Definition at line 336 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), wangle::ServerBootstrap< DefaultPipeline >::getSockets(), wangle::ServerBootstrap< DefaultPipeline >::group(), and wangle::ServerBootstrap< DefaultPipeline >::stop().

template<typename Pipeline = wangle::DefaultPipeline>

std::unique_ptr<folly::Baton<> > wangle::ServerBootstrap< Pipeline >::stopBaton_

private

Initial value:

{
    std::make_unique<folly::Baton<>>()}

Definition at line 350 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::stop(), and wangle::ServerBootstrap< DefaultPipeline >::waitForStop().

template<typename Pipeline = wangle::DefaultPipeline>

bool wangle::ServerBootstrap< Pipeline >::stopped_ {false}

private

Definition at line 352 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::stop(), and wangle::ServerBootstrap< DefaultPipeline >::waitForStop().

template<typename Pipeline = wangle::DefaultPipeline>

std::shared_ptr<ServerWorkerPool> wangle::ServerBootstrap< Pipeline >::workerFactory_

private

Definition at line 335 of file ServerBootstrap.h.

Referenced by wangle::ServerBootstrap< DefaultPipeline >::bind(), wangle::ServerBootstrap< DefaultPipeline >::bindImpl(), wangle::ServerBootstrap< DefaultPipeline >::forEachWorker(), and wangle::ServerBootstrap< DefaultPipeline >::group().

---

The documentation for this class was generated from the following file:

• proxygen/wangle/wangle/bootstrap/ServerBootstrap.h