Refactor TcpServer I/O loop logic. (#239)

Author: hwc0919

trantor/net/EventLoop.cc

@@ -241,7 +241,8 @@ void EventLoop::loop()
     catch (std::exception &e)
     {
         LOG_WARN << "Exception thrown from event loop, rethrowing after "
-                    "running functions on quit";
+                    "running functions on quit: "
+                 << e.what();
         loopException = std::current_exception();
     }
 

trantor/net/TcpServer.cc

@@ -12,31 +12,33 @@
  *
  */
 
-#include "Acceptor.h"
-#include "inner/TcpConnectionImpl.h"
 #include <trantor/net/TcpServer.h>
 #include <trantor/utils/Logger.h>
 #include <functional>
 #include <vector>
+#include "Acceptor.h"
+#include "inner/TcpConnectionImpl.h"
 using namespace trantor;
 using namespace std::placeholders;
 
 TcpServer::TcpServer(EventLoop *loop,
                      const InetAddress &address,
-                     const std::string &name,
+                     std::string name,
                      bool reUseAddr,
                      bool reUsePort)
     : loop_(loop),
       acceptorPtr_(new Acceptor(loop, address, reUseAddr, reUsePort)),
-      serverName_(name),
+      serverName_(std::move(name)),
       recvMessageCallback_([](const TcpConnectionPtr &, MsgBuffer *buffer) {
           LOG_ERROR << "unhandled recv message [" << buffer->readableBytes()
                     << " bytes]";
           buffer->retrieveAll();
-      })
+      }),
+      ioLoops_({loop}),
+      numIoLoops_(1)
 {
     acceptorPtr_->setNewConnectionCallback(
-        std::bind(&TcpServer::newConnection, this, _1, _2));
+        [this](int fd, const InetAddress &peer) { newConnection(fd, peer); });
 }
 
 TcpServer::~TcpServer()
@@ -49,21 +51,12 @@ void TcpServer::newConnection(int sockfd, const InetAddress &peer)
 {
     LOG_TRACE << "new connection:fd=" << sockfd
               << " address=" << peer.toIpPort();
-    // test code for blocking or nonblocking
-    //    std::vector<char> str(1024*1024*100);
-    //    for(int i=0;i<str.size();i++)
-    //        str[i]='A';
-    //    LOG_TRACE<<"vector size:"<<str.size();
-    //    size_t n=write(sockfd,&str[0],str.size());
-    //    LOG_TRACE<<"write "<<n<<" bytes";
     loop_->assertInLoopThread();
-    EventLoop *ioLoop = NULL;
-    if (loopPoolPtr_ && loopPoolPtr_->size() > 0)
+    EventLoop *ioLoop = ioLoops_[nextLoopIdx_];
+    if (++nextLoopIdx_ >= numIoLoops_)
     {
-        ioLoop = loopPoolPtr_->getNextLoop();
+        nextLoopIdx_ = 0;
     }
-    if (ioLoop == NULL)
-        ioLoop = loop_;
     std::shared_ptr<TcpConnectionImpl> newPtr;
     if (sslCtxPtr_)
     {
@@ -102,7 +95,10 @@ void TcpServer::newConnection(int sockfd, const InetAddress &peer)
             if (writeCompleteCallback_)
                 writeCompleteCallback_(connectionPtr);
         });
-    newPtr->setCloseCallback(std::bind(&TcpServer::connectionClosed, this, _1));
+
+    newPtr->setCloseCallback([this](const TcpConnectionPtr &closeConnPtr) {
+        connectionClosed(closeConnPtr);
+    });
     connSet_.insert(newPtr);
     newPtr->connectEstablished();
 }
@@ -114,29 +110,15 @@ void TcpServer::start()
         started_ = true;
         if (idleTimeout_ > 0)
         {
-            timingWheelMap_[loop_] =
-                std::make_shared<TimingWheel>(loop_,
-                                              idleTimeout_,
-                                              1.0F,
-                                              idleTimeout_ < 500
-                                                  ? idleTimeout_ + 1
-                                                  : 100);
-            if (loopPoolPtr_)
+            for (EventLoop *loop : ioLoops_)
             {
-                auto loopNum = loopPoolPtr_->size();
-                while (loopNum > 0)
-                {
-                    // LOG_TRACE << "new Wheel loopNum=" << loopNum;
-                    auto poolLoop = loopPoolPtr_->getNextLoop();
-                    timingWheelMap_[poolLoop] =
-                        std::make_shared<TimingWheel>(poolLoop,
-                                                      idleTimeout_,
-                                                      1.0F,
-                                                      idleTimeout_ < 500
-                                                          ? idleTimeout_ + 1
-                                                          : 100);
-                    --loopNum;
-                }
+                timingWheelMap_[loop] =
+                    std::make_shared<TimingWheel>(loop,
+                                                  idleTimeout_,
+                                                  1.0F,
+                                                  idleTimeout_ < 500
+                                                      ? idleTimeout_ + 1
+                                                      : 100);
             }
         }
         LOG_TRACE << "map size=" << timingWheelMap_.size();
@@ -156,7 +138,7 @@ void TcpServer::stop()
         {
             connPtrs.push_back(conn);
         }
-        for (auto connection : connPtrs)
+        for (auto &connection : connPtrs)
         {
             connection->forceClose();
         }
@@ -173,7 +155,7 @@ void TcpServer::stop()
             {
                 connPtrs.push_back(conn);
             }
-            for (auto connection : connPtrs)
+            for (auto &connection : connPtrs)
             {
                 connection->forceClose();
             }
@@ -223,7 +205,7 @@ void TcpServer::connectionClosed(const TcpConnectionPtr &connectionPtr)
     }
 }
 
-const std::string TcpServer::ipPort() const
+std::string TcpServer::ipPort() const
 {
     return acceptorPtr_->addr().toIpPort();
 }

trantor/net/TcpServer.h

@@ -13,18 +13,19 @@
  */
 
 #pragma once
-#include <trantor/net/callbacks.h>
-#include <trantor/utils/NonCopyable.h>
-#include <trantor/utils/Logger.h>
+#include <trantor/exports.h>
 #include <trantor/net/EventLoopThreadPool.h>
 #include <trantor/net/InetAddress.h>
 #include <trantor/net/TcpConnection.h>
+#include <trantor/net/callbacks.h>
+#include <trantor/utils/Logger.h>
+#include <trantor/utils/NonCopyable.h>
 #include <trantor/utils/TimingWheel.h>
-#include <trantor/exports.h>
-#include <string>
+#include <csignal>
 #include <memory>
 #include <set>
-#include <signal.h>
+#include <string>
+
 namespace trantor
 {
 class Acceptor;
@@ -48,7 +49,7 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
      */
     TcpServer(EventLoop *loop,
               const InetAddress &address,
-              const std::string &name,
+              std::string name,
               bool reUseAddr = true,
               bool reUsePort = true);
     ~TcpServer();
@@ -68,6 +69,7 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
     /**
      * @brief Set the number of event loops in which the I/O of connections to
      * the server is handled.
+     * An EventLoopThreadPool is created and managed by TcpServer.
      *
      * @param num
      */
@@ -76,11 +78,14 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
         assert(!started_);
         loopPoolPtr_ = std::make_shared<EventLoopThreadPool>(num);
         loopPoolPtr_->start();
+        ioLoops_ = loopPoolPtr_->getLoops();
+        numIoLoops_ = ioLoops_.size();
     }
 
     /**
      * @brief Set the event loops pool in which the I/O of connections to
      * the server is handled.
+     * A shared_ptr of EventLoopThreadPool is copied.
      *
      * @param pool
      */
@@ -89,7 +94,26 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
         assert(pool->size() > 0);
         assert(!started_);
         loopPoolPtr_ = pool;
-        loopPoolPtr_->start();
+        loopPoolPtr_->start();  // TODO: should not start by TcpServer
+        ioLoops_ = loopPoolPtr_->getLoops();
+        numIoLoops_ = ioLoops_.size();
+    }
+
+    /**
+     * @brief Set the event loops in which the I/O of connections to
+     * the server is handled.
+     * The loops are managed by caller. Caller should ensure that ioLoops
+     * lives longer than TcpServer.
+     *
+     * @param ioLoops
+     */
+    void setIoLoops(const std::vector<trantor::EventLoop *> &ioLoops)
+    {
+        assert(!ioLoops.empty());
+        assert(!started_);
+        ioLoops_ = ioLoops;
+        numIoLoops_ = ioLoops_.size();
+        loopPoolPtr_.reset();
     }
 
     /**
@@ -152,7 +176,7 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
      *
      * @return const std::string
      */
-    const std::string ipPort() const;
+    std::string ipPort() const;
 
     /**
      * @brief Get the address of the server.
@@ -178,7 +202,7 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
      */
     std::vector<EventLoop *> getIoLoops() const
     {
-        return loopPoolPtr_->getLoops();
+        return ioLoops_;
     }
 
     /**
@@ -215,10 +239,12 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
                    const std::string &caPath = "");
 
   private:
-    EventLoop *loop_;
     void handleCloseInLoop(const TcpConnectionPtr &connectionPtr);
-    std::unique_ptr<Acceptor> acceptorPtr_;
     void newConnection(int fd, const InetAddress &peer);
+    void connectionClosed(const TcpConnectionPtr &connectionPtr);
+
+    EventLoop *loop_;
+    std::unique_ptr<Acceptor> acceptorPtr_;
     std::string serverName_;
     std::set<TcpConnectionPtr> connSet_;
 
@@ -228,8 +254,18 @@ class TRANTOR_EXPORT TcpServer : NonCopyable
 
     size_t idleTimeout_{0};
     std::map<EventLoop *, std::shared_ptr<TimingWheel>> timingWheelMap_;
-    void connectionClosed(const TcpConnectionPtr &connectionPtr);
+
+    // `loopPoolPtr_` may and may not hold the internal thread pool.
+    // We should not access it directly in codes.
+    // Instead, we should use its delegation variable `ioLoops_`.
     std::shared_ptr<EventLoopThreadPool> loopPoolPtr_;
+    // If one of `setIoLoopNum()`, `setIoLoopThreadPool()` and `setIoLoops()` is
+    // called, `ioLoops_` will hold the loops passed in.
+    // Otherwise, it should contain only one element, which is `loop_`.
+    std::vector<EventLoop *> ioLoops_;
+    size_t nextLoopIdx_{0};
+    size_t numIoLoops_{0};
+
 #ifndef _WIN32
     class IgnoreSigPipe
     {