修复管道无效导致poller线程cpu占用太高问题 (ZLMediaKit#163)

xiongguangjie · Jun 10, 2023 · e4744a0 · e4744a0
1 parent 67ea918
commit e4744a0
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 19 deletions.
diff --git a/src/Poller/EventPoller.cpp b/src/Poller/EventPoller.cpp
@@ -49,11 +49,19 @@ EventPoller &EventPoller::Instance() {
     return *(EventPollerPool::Instance().getFirstPoller());
 }
 
+void EventPoller::addEventPipe() {
+    SockUtil::setNoBlocked(_pipe.readFD());
+    SockUtil::setNoBlocked(_pipe.writeFD());
+
+    // 添加内部管道事件
+    if (addEvent(_pipe.readFD(), EventPoller::Event_Read, [this](int event) { onPipeEvent(); }) == -1) {
+        throw std::runtime_error("Add pipe fd to poller failed");
+    }
+}
+
 EventPoller::EventPoller(std::string name, ThreadPool::Priority priority) {
     _name = std::move(name);
     _priority = priority;
-    SockUtil::setNoBlocked(_pipe.readFD());
-    SockUtil::setNoBlocked(_pipe.writeFD());
 
 #if defined(HAS_EPOLL)
     _epoll_fd = epoll_create(EPOLL_SIZE);
@@ -64,11 +72,7 @@ EventPoller::EventPoller(std::string name, ThreadPool::Priority priority) {
 #endif //HAS_EPOLL
     _logger = Logger::Instance().shared_from_this();
     _loop_thread_id = this_thread::get_id();
-
-    //添加内部管道事件
-    if (addEvent(_pipe.readFD(), Event_Read, [this](int event) { onPipeEvent(); }) == -1) {
-        throw std::runtime_error("Add pipe fd to poller failed");
-    }
+    addEventPipe();
 }
 
 void EventPoller::shutdown() {
@@ -221,12 +225,20 @@ bool EventPoller::isCurrentThread() {
 inline void EventPoller::onPipeEvent() {
     char buf[1024];
     int err = 0;
-    do {
-        if (_pipe.read(buf, sizeof(buf)) > 0) {
+    while (true) {
+        if ((err = _pipe.read(buf, sizeof(buf))) > 0) {
+            // 读到管道数据,继续读,直到读空为止
             continue;
         }
-        err = get_uv_error(true);
-    } while (err == UV_EINTR);
+        if (err == 0 || get_uv_error(true) != UV_EAGAIN) {
+            // 收到eof或非EAGAIN(无更多数据)错误,说明管道无效了,重新打开管道
+            ErrorL << "Invalid pipe fd of event poller, reopen it";
+            delEvent(_pipe.readFD());
+            _pipe.reOpen();
+            addEventPipe();
+        }
+        break;
+    }
 
     decltype(_list_task) _list_swap;
     {

diff --git a/src/Poller/EventPoller.h b/src/Poller/EventPoller.h
@@ -179,6 +179,11 @@ class EventPoller : public TaskExecutor, public AnyStorage, public std::enable_s
      */
     uint64_t getMinDelay();
 
+    /**
+     * 添加管道监听事件
+     */
+    void addEventPipe();
+
 private:
     class ExitException : public std::exception {};
 

diff --git a/src/Poller/PipeWrap.cpp b/src/Poller/PipeWrap.cpp
@@ -31,6 +31,11 @@ using namespace std;
 namespace toolkit {
 
 PipeWrap::PipeWrap() {
+    reOpen();
+}
+
+void PipeWrap::reOpen() {
+    clearFD();
 #if defined(_WIN32)
     const char *localip = SockUtil::support_ipv6() ? "::1" : "127.0.0.1";
     auto listener_fd = SockUtil::listen(0, localip);
@@ -48,7 +53,7 @@ PipeWrap::PipeWrap() {
     if (pipe(_pipe_fd) == -1) {
         throw runtime_error(StrPrinter << "Create posix pipe failed: " << get_uv_errmsg());
     }
-#endif // defined(_WIN32)	
+#endif // defined(_WIN32)
     SockUtil::setNoBlocked(_pipe_fd[0], true);
     SockUtil::setNoBlocked(_pipe_fd[1], false);
     SockUtil::setCloExec(_pipe_fd[0]);

diff --git a/src/Poller/PipeWrap.h b/src/Poller/PipeWrap.h
@@ -19,18 +19,15 @@ class PipeWrap {
     ~PipeWrap();
     int write(const void *buf, int n);
     int read(void *buf, int n);
-    int readFD() const {
-        return _pipe_fd[0];
-    }
-    int writeFD() const {
-        return _pipe_fd[1];
-    }
+    int readFD() const { return _pipe_fd[0]; }
+    int writeFD() const { return _pipe_fd[1]; }
+    void reOpen();
 
 private:
     void clearFD();
 
 private:
-    int _pipe_fd[2] = {-1, -1};
+    int _pipe_fd[2] = { -1, -1 };
 };
 
 } /* namespace toolkit */