Tcp/Udp服务器支持单线程模式

ZLMediaKit · Jun 21, 2024 · 79c10fe · 79c10fe
1 parent 5144e2a
commit 79c10fe
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 45 deletions.
diff --git a/src/Network/TcpServer.cpp b/src/Network/TcpServer.cpp
@@ -20,6 +20,7 @@ INSTANCE_IMP(SessionMap)
 StatisticImp(TcpServer)
 
 TcpServer::TcpServer(const EventPoller::Ptr &poller) : Server(poller) {
+    _multi_poller = !poller;
     setOnCreateSocket(nullptr);
 }
 
@@ -82,7 +83,7 @@ TcpServer::Ptr TcpServer::onCreatServer(const EventPoller::Ptr &poller) {
 Socket::Ptr TcpServer::onBeforeAcceptConnection(const EventPoller::Ptr &poller) {
     assert(_poller->isCurrentThread());
     //此处改成自定义获取poller对象，防止负载不均衡
-    return createSocket(EventPollerPool::Instance().getPoller(false));
+    return createSocket(_multi_poller ? EventPollerPool::Instance().getPoller(false) : _poller);
 }
 
 void TcpServer::cloneFrom(const TcpServer &that) {
@@ -191,19 +192,21 @@ void TcpServer::start_l(uint16_t port, const std::string &host, uint32_t backlog
         return true;
     }, _poller);
 
-    EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
-        EventPoller::Ptr poller = static_pointer_cast<EventPoller>(executor);
-        if (poller == _poller) {
-            return;
-        }
-        auto &serverRef = _cloned_server[poller.get()];
-        if (!serverRef) {
-            serverRef = onCreatServer(poller);
-        }
-        if (serverRef) {
-            serverRef->cloneFrom(*this);
-        }
-    });
+    if (_multi_poller) {
+        EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
+            EventPoller::Ptr poller = static_pointer_cast<EventPoller>(executor);
+            if (poller == _poller) {
+                return;
+            }
+            auto &serverRef = _cloned_server[poller.get()];
+            if (!serverRef) {
+                serverRef = onCreatServer(poller);
+            }
+            if (serverRef) {
+                serverRef->cloneFrom(*this);
+            }
+        });
+    }
 
     if (!_socket->listen(port, host.c_str(), backlog)) {
         // 创建tcp监听失败，可能是由于端口占用或权限问题

diff --git a/src/Network/TcpServer.h b/src/Network/TcpServer.h
@@ -92,6 +92,7 @@ class TcpServer : public Server {
     void setupEvent();
 
 private:
+    bool _multi_poller;
     bool _is_on_manager = false;
     bool _main_server = true;
     std::weak_ptr<TcpServer> _parent;

diff --git a/src/Network/UdpServer.cpp b/src/Network/UdpServer.cpp
@@ -45,6 +45,7 @@ static UdpServer::PeerIdType makeSockId(sockaddr *addr, int) {
 }
 
 UdpServer::UdpServer(const EventPoller::Ptr &poller) : Server(poller) {
+    _multi_poller = !poller;
     setOnCreateSocket(nullptr);
 }
 
@@ -87,20 +88,22 @@ void UdpServer::start_l(uint16_t port, const std::string &host) {
         return false;
     }, _poller);
 
-    //clone server至不同线程，让udp server支持多线程
-    EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
-        auto poller = std::static_pointer_cast<EventPoller>(executor);
-        if (poller == _poller) {
-            return;
-        }
-        auto &serverRef = _cloned_server[poller.get()];
-        if (!serverRef) {
-            serverRef = onCreatServer(poller);
-        }
-        if (serverRef) {
-            serverRef->cloneFrom(*this);
-        }
-    });
+    if (_multi_poller) {
+        // clone server至不同线程，让udp server支持多线程
+        EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
+            auto poller = std::static_pointer_cast<EventPoller>(executor);
+            if (poller == _poller) {
+                return;
+            }
+            auto &serverRef = _cloned_server[poller.get()];
+            if (!serverRef) {
+                serverRef = onCreatServer(poller);
+            }
+            if (serverRef) {
+                serverRef->cloneFrom(*this);
+            }
+        });
+    }
 
     if (!_socket->bindUdpSock(port, host.c_str())) {
         // udp 绑定端口失败, 可能是由于端口占用或权限问题
@@ -193,24 +196,28 @@ void UdpServer::onManagerSession() {
         //拷贝map，防止遍历时移除对象
         copy_map = std::make_shared<std::unordered_map<PeerIdType, SessionHelper::Ptr> >(*_session_map);
     }
-    EventPollerPool::Instance().for_each([copy_map](const TaskExecutor::Ptr &executor) {
-        auto poller = std::static_pointer_cast<EventPoller>(executor);
-        poller->async([copy_map]() {
-            for (auto &pr : *copy_map) {
-                auto &session = pr.second->session();
-                if (!session->getPoller()->isCurrentThread()) {
-                    //该session不归属该poller管理
-                    continue;
-                }
-                try {
-                    // UDP 会话需要处理超时
-                    session->onManager();
-                } catch (exception &ex) {
-                    WarnL << "Exception occurred when emit onManager: " << ex.what();
-                }
+    auto lam = [copy_map]() {
+        for (auto &pr : *copy_map) {
+            auto &session = pr.second->session();
+            if (!session->getPoller()->isCurrentThread()) {
+                // 该session不归属该poller管理
+                continue;
             }
+            try {
+                // UDP 会话需要处理超时
+                session->onManager();
+            } catch (exception &ex) {
+                WarnL << "Exception occurred when emit onManager: " << ex.what();
+            }
+        }
+    };
+    if (_multi_poller){
+        EventPollerPool::Instance().for_each([lam](const TaskExecutor::Ptr &executor) {
+            std::static_pointer_cast<EventPoller>(executor)->async(lam);
         });
-    });
+    } else {
+        lam();
+    }
 }
 
 SessionHelper::Ptr UdpServer::getOrCreateSession(const UdpServer::PeerIdType &id, const Buffer::Ptr &buf, sockaddr *addr, int addr_len, bool &is_new) {
@@ -228,7 +235,7 @@ SessionHelper::Ptr UdpServer::getOrCreateSession(const UdpServer::PeerIdType &id
 
 SessionHelper::Ptr UdpServer::createSession(const PeerIdType &id, const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
     // 此处改成自定义获取poller对象，防止负载不均衡
-    auto socket = createSocket(EventPollerPool::Instance().getPoller(false), buf, addr, addr_len);
+    auto socket = createSocket(_multi_poller ? EventPollerPool::Instance().getPoller(false) : _poller, buf, addr, addr_len);
     if (!socket) {
         //创建socket失败，本次onRead事件收到的数据直接丢弃
         return nullptr;

diff --git a/src/Network/UdpServer.h b/src/Network/UdpServer.h
@@ -108,6 +108,7 @@ class UdpServer : public Server {
 
 private:
     bool _cloned = false;
+    bool _multi_poller;
     Socket::Ptr _socket;
     std::shared_ptr<Timer> _timer;
     onCreateSocket _on_create_socket;