Add ConnectionAcceptor::stop() (#406)

Its behavior is to have the server stop listening, whatever server may be inside
the ConnectionAcceptor...  This is very helpful towards having sane shutdown
semantics.

Moves us closer to having ~RSocketServer not race as hard.

Author: alexmalyshev

experimental/rsocket-src/RSocketServer.cpp

@@ -47,6 +47,14 @@ RSocketServer::RSocketServer(
       acceptor_(ProtocolVersion::Unknown) {}
 
 RSocketServer::~RSocketServer() {
+  // Stop accepting new connections.
+  lazyAcceptor_->stop();
+
+  // FIXME(alexanderm): This is where we /should/ close the FrameTransports
+  // sitting around in the ServerConnectionAcceptor, but we can't yet...
+
+  // Asynchronously close all existing ReactiveSockets.  If there are none, then
+  // we can do an early exit.
   {
     auto locked = sockets_.lock();
     if (locked->empty()) {
@@ -61,35 +69,34 @@ RSocketServer::~RSocketServer() {
     }
   }
 
+  // Wait for all ReactiveSockets to close.
   shutdown_->wait();
   DCHECK(sockets_.lock()->empty());
+
+  // All requests are fully finished, worker threads can be safely killed off.
 }
 
 void RSocketServer::start(OnAccept onAccept) {
   if (connectionHandler_) {
     throw std::runtime_error("RSocketServer::start() already called.");
   }
 
-  LOG(INFO) << "RSocketServer => initialize connection acceptor on start";
+  LOG(INFO) << "Initializing connection acceptor on start";
 
-  LOG(INFO) << "RSocketServer => initialize connection acceptor on start";
   connectionHandler_ =
       std::make_unique<RSocketServerConnectionHandler>(this, onAccept);
 
   lazyAcceptor_
       ->start([this](
                   std::unique_ptr<DuplexConnection> conn,
                   folly::Executor& executor) {
-        LOG(INFO) << "RSocketServer => received new connection";
+        LOG(INFO) << "Going to accept duplex connection";
 
-        LOG(INFO) << "RSocketServer => going to accept duplex connection";
-        // the callbacks above are wired up, now accept the connection
         // FIXME(alexanderm): This isn't thread safe
         acceptor_.accept(std::move(conn), connectionHandler_);
       })
       .onError([](const folly::exception_wrapper& ex) {
-        LOG(FATAL) << "RSocketServer => failed to start HttpAcceptor: "
-                   << ex.what();
+        LOG(FATAL) << "Failed to start ConnectionAcceptor: " << ex.what();
       });
 }
 

experimental/rsocket-src/transports/TcpConnectionAcceptor.cpp

@@ -60,15 +60,13 @@ TcpConnectionAcceptor::TcpConnectionAcceptor(Options options)
     : options_(std::move(options)) {}
 
 TcpConnectionAcceptor::~TcpConnectionAcceptor() {
-  LOG(INFO) << "Shutting down TCP listener";
-
-  // Need to terminate ServerSocket before the EventBase.  The socket will
-  // access the EventBase in its destructor.
-  serverThread_->getEventBase()->runInEventBaseThread(
-      [this] { serverSocket_.reset(); });
-  serverThread_.reset();
+  if (serverThread_) {
+    stop();
+  }
 }
 
+////////////////////////////////////////////////////////////////////////////////
+
 folly::Future<folly::Unit> TcpConnectionAcceptor::start(
     std::function<void(std::unique_ptr<DuplexConnection>, folly::EventBase&)>
         acceptor) {
@@ -116,4 +114,12 @@ folly::Future<folly::Unit> TcpConnectionAcceptor::start(
   LOG(INFO) << "ConnectionAcceptor => leave start";
   return folly::unit;
 }
+
+void TcpConnectionAcceptor::stop() {
+  LOG(INFO) << "Shutting down TCP listener";
+
+  serverThread_->getEventBase()->runInEventBaseThread(
+      [this] { serverSocket_.reset(); });
+  serverThread_.reset();
+}
 }

experimental/rsocket/ConnectionAcceptor.h

@@ -20,6 +20,8 @@ using OnDuplexConnectionAccept = std::function<
  *
  * Built-in implementations can be found in rsocket/transports/, such as
  * rsocket/transports/TcpConnectionAcceptor.h
+ *
+ * TODO: Add way of specifying number of worker threads.
  */
 class ConnectionAcceptor {
  public:
@@ -43,6 +45,13 @@ class ConnectionAcceptor {
    */
   virtual folly::Future<folly::Unit> start(
       OnDuplexConnectionAccept onAccept) = 0;
-  // TODO need to add numThreads option (either overload or arg with default=1)
+
+  /**
+   * Stop listening for new connections.
+   *
+   * This can only be called once.  Must be called in or before
+   * the implementation's destructor.  Must be synchronous.
+   */
+  virtual void stop() = 0;
 };
 }

experimental/rsocket/transports/TcpConnectionAcceptor.h

@@ -40,14 +40,17 @@ class TcpConnectionAcceptor : public ConnectionAcceptor {
 
   /**
    * Bind an AsyncServerSocket and start accepting TCP connections.
-   *
-   * This can only be called once.
    */
   folly::Future<folly::Unit> start(
       std::function<void(
           std::unique_ptr<reactivesocket::DuplexConnection>,
           folly::EventBase&)>) override;
 
+  /**
+   * Shutdown the AsyncServerSocket and associated listener thread.
+   */
+  void stop() override;
+
  private:
   class SocketCallback;