feat: 封装了轮询非阻塞 EpollLoop 并通过测试

CMakeLists.txt

@@ -10,5 +10,5 @@ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -static -g -O2")
 
 add_subdirectory(${PROJECT_SOURCE_DIR}/include/ouc_server)
 
-add_executable(test "${PROJECT_SOURCE_DIR}/examples/test_thread_pool.cpp")
+add_executable(test "${PROJECT_SOURCE_DIR}/examples/epoll_tcp_loop.cpp")
 target_link_libraries(test PRIVATE ouc_server_lib)

examples/epoll_tcp_loop.cpp

@@ -1,6 +1,7 @@
 #include <epoll/epoll_loop.hpp>
 
 #include <fcntl.h>
+#include <string.h>
 #include <iostream>
 
 int set_nonblock(int fd)
@@ -42,37 +43,45 @@ int main()
 
     puts("Server listening on port 8080...");
 
-    EpollLoop loop;
+    EpollLoop loop(4);
 
     loop.add_fd(
         sock_fd,
         EPOLLIN,
-        [&]()
+        [&](int fd)
         {
-        while (true) {
-            int client_fd = accept(sock_fd, nullptr, nullptr);
-            if (client_fd < 0) break;
+        while (true)
+        {
+            int client_fd = accept(fd, nullptr, nullptr);
+            if (client_fd < 0)
+            {
+                if (errno == EAGAIN || errno == EWOULDBLOCK) 
+                    break;
+                perror("accept");
+                break;
+            }
             set_nonblock(client_fd);
-            std::cout << "New client: " << client_fd << "\n";
 
             loop.add_fd(
                 client_fd, 
                 EPOLLIN, 
-                [&, client_fd]() 
+                [](int cfd)
                 {
-                char buf[1024];
-                ssize_t r = read(client_fd, buf, sizeof(buf));
-                if (r <= 0) {
-                    std::cout << "Client " << client_fd << " disconnected\n";
-                    loop.remove_fd(client_fd);
-                    close(client_fd);
+                    char buf[4096];
+                    ssize_t n = read(cfd, buf, sizeof(buf));
+                    write(cfd, buf, n);
+                    if ((!memcmp("exit", buf, 4))) 
+                    {
+                        close(cfd);
                         return;
                     }
-                std::string msg(buf, r);
-                std::cout << "Recv: " << msg;
-                write(client_fd, buf, r);
+                    std::string data(buf, n);
+                    std::cout<< "recv:" << data;
             });
     } });
 
-    loop.run();
+    while (true)
+    {
+        loop.poll(10);
+    }
 }

include/ouc_server/epoll/epoll_loop.cpp

@@ -11,9 +11,12 @@ namespace ouc_server
     namespace epoll
     {
 
-        EpollLoop::EpollLoop()
+        EpollLoop::EpollLoop(size_t n)
+            : pool(n)
         {
             epoll_fd = epoll_create1(0);
+            if (epoll_fd < 0)
+                perror("epoll_create1");
         }
 
         EpollLoop::~EpollLoop()
@@ -21,24 +24,27 @@ namespace ouc_server
             close(epoll_fd);
         }
 
-        void EpollLoop::run(int timeout_ms)
+        void EpollLoop::poll(const int timeout_ms, const int MAX_EVENTS)
         {
-            std::vector<struct epoll_event> events(64);
-            while (true)
-            {
-                int nfds = epoll_wait(epoll_fd, events.data(), events.size(), timeout_ms);
+            epoll_event events[MAX_EVENTS];
+
+            int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, timeout_ms);
             if (nfds < 0)
             {
+                if (errno == EINTR)
+                    return;
                 perror("epoll_wait");
-                    break;
+                return;
             }
+
             for (int i = 0; i < nfds; ++i)
             {
                 int fd = events[i].data.fd;
-                    if (events_map.count(fd))
+                if (callbacks.count(fd))
                 {
-                        events_map[fd].callback();
-                    }
+                    pool.sumbit(
+                        [ev = callbacks[fd], fd]()
+                        { ev.callback(fd); });
                 }
             }
         }
@@ -47,7 +53,7 @@ namespace ouc_server
         {
             auto ev = pack_event(fd, event_flags);
             int code = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev);
-            events_map[fd] = Event{fd, event_flags, callback};
+            callbacks[fd] = Event{fd, event_flags, std::move(callback)};
             return code == 0;
         }
 
@@ -55,14 +61,14 @@ namespace ouc_server
         {
             auto ev = pack_event(fd, event_flags);
             int code = epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ev);
-            events_map[fd].events = event_flags;
+            callbacks[fd].events = event_flags;
             return code == 0;
         }
 
         bool EpollLoop::remove_fd(int fd)
         {
             int code = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
-            events_map.erase(fd);
+            callbacks.erase(fd);
             return code == 0;
         }
 

include/ouc_server/epoll/epoll_loop.hpp

@@ -10,11 +10,13 @@
 #include <sys/epoll.h>
 #include <netinet/in.h>
 
+#include <utils/thread_pool.hpp>
+
 namespace ouc_server
 {
     namespace epoll
     {
-        using EpollCallback = std::function<void()>;
+        using EpollCallback = std::function<void(int)>;
 
         struct Event
         {
@@ -25,18 +27,19 @@ namespace ouc_server
 
         class EpollLoop
         {
-        public:
         private:
             int epoll_fd;
-            std::unordered_map<int, Event> events_map;
+            std::unordered_map<int, Event> callbacks;
+
+            ouc_server::utils::ThreadPool pool;
 
         public:
-            EpollLoop();
+            EpollLoop(size_t = 64);
 
             ~EpollLoop();
 
         public:
-            void run(int = -1);
+            void poll(const int = 0, const int = 64);
 
             bool add_fd(int, uint32_t, EpollCallback);
             bool modify_fd(int, uint32_t);