merge:1.8.7

Author: hevake

examples/event/09_named_pipe/Makefile

@@ -0,0 +1,32 @@
+#
+#     .============.
+#    //  M A K E  / \
+#   //  C++ DEV  /   \
+#  //  E A S Y  /  \/ \
+# ++ ----------.  \/\  .
+#  \\     \     \ /\  /
+#   \\     \     \   /
+#    \\     \     \ /
+#     -============'
+#
+# Copyright (c) 2024 Hevake and contributors, all rights reserved.
+#
+# This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
+# Use of this source code is governed by MIT license that can be found
+# in the LICENSE file in the root of the source tree. All contributing
+# project authors may be found in the CONTRIBUTORS.md file in the root
+# of the source tree.
+#
+
+PROJECT := example/event/09_named_pipe
+EXE_NAME := 09_named_pipe
+
+CPP_SRC_FILES := main.cpp
+
+CXXFLAGS := -DLOG_MODULE_ID='"$(EXE_NAME)"' $(CXXFLAGS)
+LDFLAGS += \
+	-ltbox_event \
+	-ltbox_base \
+	-ldl
+
+include ${TOP_DIR}/tools/exe_common.mk

examples/event/09_named_pipe/main.cpp

@@ -0,0 +1,98 @@
+/*
+ *     .============.
+ *    //  M A K E  / \
+ *   //  C++ DEV  /   \
+ *  //  E A S Y  /  \/ \
+ * ++ ----------.  \/\  .
+ *  \\     \     \ /\  /
+ *   \\     \     \   /
+ *    \\     \     \ /
+ *     -============'
+ *
+ * Copyright (c) 2024 Hevake and contributors, all rights reserved.
+ *
+ * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
+ * Use of this source code is governed by MIT license that can be found
+ * in the LICENSE file in the root of the source tree. All contributing
+ * project authors may be found in the CONTRIBUTORS.md file in the root
+ * of the source tree.
+ */
+/**
+ * 本示例，有名管道的读取
+ */
+
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include <iostream>
+#include <sstream>
+#include <cstring>
+
+#include <tbox/base/log.h>
+#include <tbox/base/log_output.h>
+#include <tbox/event/loop.h>
+#include <tbox/event/fd_event.h>
+
+using namespace std;
+using namespace tbox;
+using namespace tbox::event;
+
+void PrintUsage(const char *process_name)
+{
+    cout << "Usage:" << process_name << " your_pipefile" << endl;
+}
+
+int main(int argc, char *argv[])
+{
+    if (argc < 2) {
+        PrintUsage(argv[0]);
+        return 0;
+    }
+
+    LogOutput_Enable();
+    Loop* sp_loop = Loop::New();
+
+    const char* pipe_file = argv[1];
+    int ret = mkfifo(pipe_file, 0666);
+    if (ret != 0) {
+        LogWarn("mkfifo() ret:%d, errno:%d", ret, errno);
+    }
+    int fd = open(pipe_file, O_RDONLY|O_NDELAY);
+
+    FdEvent* sp_fd_read  = sp_loop->newFdEvent();
+    sp_fd_read->initialize(fd, FdEvent::kReadEvent | FdEvent::kHupEvent, Event::Mode::kPersist);    //! 可读与挂起事件一直有效
+    sp_fd_read->enable();
+
+    sp_fd_read->setCallback(
+        [&] (short event) {
+            //! 当管道有输入的时候
+            if (event & FdEvent::kReadEvent) {
+                char input_buff[200];
+                int rsize = read(fd, input_buff, sizeof(input_buff));
+                if (rsize > 0) {
+                    input_buff[rsize - 1] = '\0';
+                    LogInfo("read[%d]:%s", rsize, input_buff);
+                } else {
+                    LogInfo("read 0");
+                }
+            }
+
+            //! 当管道断开的时候
+            if (event & FdEvent::kHupEvent) {
+                LogNotice("hup");
+                fd = open(pipe_file, O_RDONLY|O_NDELAY);
+                sp_fd_read->initialize(fd, FdEvent::kReadEvent | FdEvent::kHupEvent, Event::Mode::kPersist);
+                sp_fd_read->enable();
+            }
+        }
+    );
+
+    sp_loop->runLoop(Loop::Mode::kForever);
+
+    delete sp_fd_read;
+    delete sp_loop;
+
+    return 0;
+}

modules/event/engines/epoll/fd_event.cpp

@@ -70,13 +70,15 @@ bool EpollFdEvent::enable()
         return true;
 
     if (events_ & kReadEvent)
-        d_->read_events.push_back(this);
+        ++d_->read_event_num;
 
     if (events_ & kWriteEvent)
-        d_->write_events.push_back(this);
+        ++d_->write_event_num;
 
     if (events_ & kExceptEvent)
-        d_->exception_events.push_back(this);
+        ++d_->except_event_num;
+
+    d_->fd_events.push_back(this);
 
     reloadEpoll();
 
@@ -89,20 +91,17 @@ bool EpollFdEvent::disable()
     if (d_ == nullptr || !is_enabled_)
         return true;
 
-    if (events_ & kReadEvent) {
-        auto iter = std::find(d_->read_events.begin(), d_->read_events.end(), this);
-        d_->read_events.erase(iter);
-    }
+    if (events_ & kReadEvent)
+        --d_->read_event_num;
 
-    if (events_ & kWriteEvent) {
-        auto iter = std::find(d_->write_events.begin(), d_->write_events.end(), this);
-        d_->write_events.erase(iter);
-    }
+    if (events_ & kWriteEvent)
+        --d_->write_event_num;
 
-    if (events_ & kExceptEvent) {
-        auto iter = std::find(d_->exception_events.begin(), d_->exception_events.end(), this);
-        d_->exception_events.erase(iter);
-    }
+    if (events_ & kExceptEvent)
+        --d_->except_event_num;
+
+    auto iter = std::find(d_->fd_events.begin(), d_->fd_events.end(), this);
+    d_->fd_events.erase(iter);
 
     reloadEpoll();
 
@@ -121,12 +120,14 @@ void EpollFdEvent::reloadEpoll()
     uint32_t old_events = d_->ev.events;
     uint32_t new_events = 0;
 
-    if (!d_->write_events.empty())
+    if (d_->write_event_num > 0)
         new_events |= EPOLLOUT;
-    if (!d_->read_events.empty())
+
+    if (d_->read_event_num > 0)
         new_events |= EPOLLIN;
-    if (!d_->exception_events.empty())
-        new_events |= (EPOLLHUP | EPOLLERR);
+
+    if (d_->except_event_num > 0)
+        new_events |= EPOLLERR;
 
     d_->ev.events = new_events;
 
@@ -141,45 +142,61 @@ void EpollFdEvent::reloadEpoll()
     }
 }
 
-void EpollFdEvent::OnEventCallback(int fd, uint32_t events, void *obj)
+void EpollFdEvent::OnEventCallback(uint32_t events, void *obj)
 {
     EpollFdSharedData *d = static_cast<EpollFdSharedData*>(obj);
 
+    short tbox_events = 0;
     if (events & EPOLLIN) {
         events &= ~EPOLLIN;
-        for (EpollFdEvent *event : d->read_events)
-            event->onEvent(kReadEvent);
+        tbox_events |= kReadEvent;
     }
 
     if (events & EPOLLOUT) {
         events &= ~EPOLLOUT;
-        for (EpollFdEvent *event : d->write_events)
-            event->onEvent(kWriteEvent);
+        tbox_events |= kWriteEvent;
     }
 
-    if (events & (EPOLLHUP | EPOLLERR)) {
-        events &= ~(EPOLLHUP | EPOLLERR);
-        for (EpollFdEvent *event : d->exception_events)
-            event->onEvent(kExceptEvent);
+    if (events & EPOLLERR) {
+        events &= ~EPOLLERR;
+        tbox_events |= kExceptEvent;
     }
 
-    if (events) {
-        LogNotice("unhandle events:%08X, fd:%d", events, fd);
+    if (events & EPOLLHUP) {
+        events &= ~EPOLLHUP;
+        tbox_events |= kHupEvent;
     }
+
+    //! 要先复制一份，因为在for中很可能会改动到d->fd_events，引起迭代器失效问题
+    auto tmp = d->fd_events;
+    for (auto event : tmp)
+        event->onEvent(tbox_events);
+
+    if (events)
+        LogWarn("unhandle events:%08X, fd:%d", events, d->fd);
 }
 
 void EpollFdEvent::onEvent(short events)
 {
-    if (is_stop_after_trigger_)
+    /**
+     * 由于EPOLLHUP会一直触发，所以无论事件有没有监听HupEvent，只要发生了EPOLLHUB事件，
+     * 对应fd所有的事件都要强制disable()。否则会导致Loop空跑问题。
+     */
+    if (events & kHupEvent)
         disable();
 
-    wp_loop_->beginEventProcess();
-    if (cb_) {
-        ++cb_level_;
-        cb_(events);
-        --cb_level_;
+    if (events_ & events) {
+        if (is_stop_after_trigger_)
+            disable();
+
+        wp_loop_->beginEventProcess();
+        if (cb_) {
+            ++cb_level_;
+            cb_(events);
+            --cb_level_;
+        }
+        wp_loop_->endEventProcess(this);
     }
-    wp_loop_->endEventProcess(this);
 }
 
 }

modules/event/engines/epoll/fd_event.h

@@ -44,7 +44,7 @@ class EpollFdEvent : public FdEvent {
     virtual Loop* getLoop() const override;
 
   public:
-    static void OnEventCallback(int fd, uint32_t events, void *obj);
+    static void OnEventCallback(uint32_t events, void *obj);
 
   protected:
     void reloadEpoll();

modules/event/engines/epoll/loop.cpp

@@ -76,7 +76,7 @@ void EpollLoop::runLoop(Mode mode)
 
         for (int i = 0; i < fds; ++i) {
             epoll_event &ev = events.at(i);
-            EpollFdEvent::OnEventCallback(ev.data.fd, ev.events, ev.data.ptr);
+            EpollFdEvent::OnEventCallback(ev.events, ev.data.ptr);
         }
 
         //handleRunInLoopFunc();
@@ -108,6 +108,7 @@ EpollFdSharedData* EpollLoop::refFdSharedData(int fd)
         TBOX_ASSERT(fd_shared_data != nullptr);
 
         ::memset(&fd_shared_data->ev, 0, sizeof(fd_shared_data->ev));
+        fd_shared_data->fd = fd;
         fd_shared_data->ev.data.ptr = static_cast<void *>(fd_shared_data);
 
         fd_data_map_.insert(std::make_pair(fd, fd_shared_data));

modules/event/engines/epoll/types.h

@@ -30,11 +30,15 @@ class EpollFdEvent;
 
 //! 同一个fd共享的数据
 struct EpollFdSharedData {
-    int ref = 0;    //! 引用计数
+    int fd = 0;     //!< 文件描述符
+    int ref = 0;    //!< 引用计数
     struct epoll_event ev;
-    std::vector<EpollFdEvent*> read_events;
-    std::vector<EpollFdEvent*> write_events;
-    std::vector<EpollFdEvent*> exception_events;
+
+    int read_event_num = 0;     //!< 监听可读事件的FdEvent个数
+    int write_event_num = 0;    //!< 监听可写事件的FdEvent个数
+    int except_event_num = 0;   //!< 监听异常事件的FdEvent个数
+
+    std::vector<EpollFdEvent*> fd_events;
 };
 
 }

modules/event/fd_event.h

@@ -30,9 +30,10 @@ namespace event {
 class FdEvent : public Event {
   public:
     enum EventTypes {
-        kReadEvent   = 0x01,
-        kWriteEvent  = 0x02,
-        kExceptEvent = 0x04,
+        kReadEvent   = 0x01,    //!< 可读事件
+        kWriteEvent  = 0x02,    //!< 可写事件
+        kExceptEvent = 0x04,    //!< 异常事件
+        kHupEvent    = 0x08,    //!< 断开事件
     };
 
     using Event::Event;

modules/event/fd_event_test.cpp

@@ -361,23 +361,23 @@ TEST(FdEvent, Exception)
 
     int run_time = 0;
 
-    EXPECT_TRUE(read_fd_event->initialize(read_fd, FdEvent::kReadEvent | FdEvent::kExceptEvent, Event::Mode::kPersist));
+    EXPECT_TRUE(read_fd_event->initialize(read_fd, FdEvent::kReadEvent | FdEvent::kHupEvent, Event::Mode::kPersist));
     read_fd_event->setCallback([&](short events){
         if (events & FdEvent::kReadEvent) {
             char data[100] = { 0};
             ssize_t len = read(read_fd, data, sizeof(data));
             EXPECT_EQ(len, sizeof(int));
         }
 
-        if (events & FdEvent::kExceptEvent) {
+        if (events & FdEvent::kHupEvent) {
             ++run_time;
             loop->exitLoop();
         }
     });
 
     read_fd_event->enable();
 
-    EXPECT_TRUE(write_fd_event->initialize(write_fd, FdEvent::kWriteEvent | FdEvent::kExceptEvent, Event::Mode::kPersist));
+    EXPECT_TRUE(write_fd_event->initialize(write_fd, FdEvent::kWriteEvent, Event::Mode::kPersist));
     write_fd_event->setCallback([&](short events){
         if (events & FdEvent::kWriteEvent) {
             int data = 0;

version.mk

@@ -1,4 +1,4 @@
 # TBOX版本号
 TBOX_VERSION_MAJOR := 1
 TBOX_VERSION_MINOR := 8
-TBOX_VERSION_REVISION := 6
+TBOX_VERSION_REVISION := 7