Refactor ReactorPoll, Fixed a crash caused by adding and removing events in the Poll event handler on the Cygwin platform. Fix GH-5760 GH-5759 GH-5758

Author: matyhtf

ext-src/swoole_pgsql.cc

@@ -25,8 +25,8 @@
 
 using swoole::Coroutine;
 using swoole::Reactor;
+using swoole::translate_events_to_poll;
 using swoole::coroutine::Socket;
-using swoole::coroutine::translate_events_to_poll;
 
 static SW_THREAD_LOCAL bool swoole_pgsql_blocking = true;
 

include/swoole_coroutine_system.h

@@ -18,7 +18,6 @@
 #pragma once
 
 #include "swoole_coroutine.h"
-#include "swoole_file.h"
 
 #include <vector>
 
@@ -39,9 +38,6 @@ struct PollSocket {
     }
 };
 
-int16_t translate_events_to_poll(int events);
-int translate_events_from_poll(int16_t events);
-
 class System {
   public:
     static void init_reactor(Reactor *reactor);

include/swoole_reactor.h

@@ -297,9 +297,7 @@ class Reactor {
             return error_handler[fd_type] ? error_handler[fd_type] : default_error_handler;
         default:
             abort();
-            break;
         }
-        return nullptr;
     }
 
     ReactorHandler get_error_handler(const FdType fd_type) const {
@@ -339,6 +337,10 @@ class Reactor {
         _socket->events = events;
     }
 
+    bool _exists(const network::Socket *_socket) {
+        return sockets_.find(_socket->fd) != sockets_.end();
+    }
+
     void _del(network::Socket *_socket) {
         _socket->events = 0;
         _socket->removed = 1;
@@ -377,6 +379,9 @@ class Reactor {
         return events & SW_EVENT_ERROR;
     }
 };
+
+int16_t translate_events_to_poll(int events);
+int translate_events_from_poll(int16_t events);
 }  // namespace swoole
 
 #define SW_REACTOR_CONTINUE                                                                                            \

src/coroutine/hook.cc

@@ -30,13 +30,13 @@
 
 using swoole::AsyncEvent;
 using swoole::Coroutine;
+using swoole::translate_events_from_poll;
+using swoole::translate_events_to_poll;
 using swoole::async::dispatch;
 using swoole::coroutine::async;
 using swoole::coroutine::PollSocket;
 using swoole::coroutine::Socket;
 using swoole::coroutine::System;
-using swoole::coroutine::translate_events_from_poll;
-using swoole::coroutine::translate_events_to_poll;
 
 #ifdef SW_USE_IOURING
 using swoole::Iouring;

src/coroutine/system.cc

@@ -379,36 +379,6 @@ static int socket_poll_error_callback(Reactor *reactor, Event *event) {
     return SW_OK;
 }
 
-int16_t translate_events_to_poll(int events) {
-    int16_t poll_events = 0;
-
-    if (events & SW_EVENT_READ) {
-        poll_events |= POLLIN;
-    }
-    if (events & SW_EVENT_WRITE) {
-        poll_events |= POLLOUT;
-    }
-
-    return poll_events;
-}
-
-int translate_events_from_poll(int16_t events) {
-    int sw_events = 0;
-
-    if (events & POLLIN) {
-        sw_events |= SW_EVENT_READ;
-    }
-    if (events & POLLOUT) {
-        sw_events |= SW_EVENT_WRITE;
-    }
-    // ignore ERR and HUP, because event is already processed at IN and OUT handler.
-    if ((((events & POLLERR) || (events & POLLHUP)) && !((events & POLLIN) || (events & POLLOUT)))) {
-        sw_events |= SW_EVENT_ERROR;
-    }
-
-    return sw_events;
-}
-
 bool System::socket_poll(std::unordered_map<int, PollSocket> &fds, double timeout) {
     if (timeout == 0) {
         struct pollfd *event_list = (struct pollfd *) sw_calloc(fds.size(), sizeof(struct pollfd));

src/reactor/base.cc

@@ -77,7 +77,7 @@ Reactor::Reactor(int max_event, Type _type) {
     switch (type_) {
 #ifdef HAVE_EPOLL
     case TYPE_EPOLL:
-        impl = make_reactor_epoll(this, max_event);
+        impl = make_reactor_poll(this, max_event);
         break;
 #endif
 #ifdef HAVE_KQUEUE

src/reactor/poll.cc

@@ -21,15 +21,12 @@
 #include "swoole_reactor.h"
 
 namespace swoole {
-
 using network::Socket;
 
 class ReactorPoll final : public ReactorImpl {
-    uint32_t max_fd_num;
-    Socket **fds_;
     pollfd *events_;
-    bool exists(int fd) const;
-    void set_events(int index, int events) const;
+    int max_events_;
+    int set_events() const;
 
   public:
     ReactorPoll(Reactor *_reactor, int max_events);
@@ -48,34 +45,30 @@ ReactorImpl *make_reactor_poll(Reactor *_reactor, int max_events) {
 }
 
 ReactorPoll::ReactorPoll(Reactor *_reactor, int max_events) : ReactorImpl(_reactor) {
-    fds_ = new Socket *[max_events];
     events_ = new pollfd[max_events];
-
-    max_fd_num = max_events;
+    max_events_ = max_events;
     reactor_->max_event_num = max_events;
 }
 
 ReactorPoll::~ReactorPoll() {
-    delete[] fds_;
     delete[] events_;
 }
 
-void ReactorPoll::set_events(const int index, const int events) const {
-    events_[index].events = 0;
-    if (Reactor::isset_read_event(events)) {
-        events_[index].events |= POLLIN;
-    }
-    if (Reactor::isset_write_event(events)) {
-        events_[index].events |= POLLOUT;
-    }
-    if (Reactor::isset_error_event(events)) {
-        events_[index].events |= POLLHUP;
+int ReactorPoll::set_events() const {
+    const auto sockets = reactor_->get_sockets();
+    int count = 0;
+    for (const auto pair : sockets) {
+        const auto _socket = pair.second;
+        events_[count].fd = _socket->fd;
+        events_[count].events = translate_events_to_poll(_socket->events);
+        events_[count].revents = 0;
+        count++;
     }
+    return count;
 }
 
 int ReactorPoll::add(Socket *socket, const int events) {
-    int fd = socket->fd;
-    if (exists(fd)) {
+    if (reactor_->_exists(socket)) {
         swoole_error_log(
             SW_LOG_WARNING,
             SW_ERROR_EVENT_ADD_FAILED,
@@ -89,45 +82,38 @@ int ReactorPoll::add(Socket *socket, const int events) {
         return SW_ERR;
     }
 
-    const int cur = reactor_->get_event_num();
-    if (reactor_->get_event_num() == max_fd_num) {
+    if (reactor_->get_event_num() == static_cast<size_t>(max_events_)) {
         swoole_error_log(
-            SW_LOG_WARNING, SW_ERROR_EVENT_ADD_FAILED, "too many sockets, the max events is %d", max_fd_num);
+            SW_LOG_WARNING, SW_ERROR_EVENT_ADD_FAILED, "too many sockets, the max events is %d", max_events_);
         swoole_print_backtrace_on_error();
         return SW_ERR;
     }
 
+    swoole_trace("fd=%d, events=%d", socket->fd, events);
     reactor_->_add(socket, events);
 
-    swoole_trace("fd=%d, events=%d", fd, events);
-
-    fds_[cur] = socket;
-    events_[cur].fd = fd;
-    set_events(cur, events);
-
     return SW_OK;
 }
 
 int ReactorPoll::set(Socket *socket, int events) {
-    swoole_trace("fd=%d, events=%d", socket->fd, events);
-
-    SW_LOOP_N(reactor_->get_event_num()) {
-        if (events_[i].fd == socket->fd) {
-            set_events(i, events);
-            reactor_->_set(socket, events);
-            return SW_OK;
-        }
+    if (!reactor_->_exists(socket)) {
+        swoole_error_log(
+            SW_LOG_WARNING,
+            SW_ERROR_SOCKET_NOT_EXISTS,
+            "[Reactor#%d] failed to set events[fd=%d, fd_type=%d, events=%d], the socket#%d has already been removed",
+            reactor_->id,
+            socket->fd,
+            socket->fd_type,
+            events,
+            socket->fd);
+        swoole_print_backtrace_on_error();
+        return SW_ERR;
     }
 
-    swoole_error_log(SW_LOG_WARNING,
-                     SW_ERROR_SOCKET_NOT_EXISTS,
-                     "[Reactor#%d] failed to set events[fd=%d, fd_type=%d, events=%d], the socket#%d is not exists",
-                     reactor_->id,
-                     socket->fd,
-                     socket->fd_type,
-                     events,
-                     socket->fd);
-    return SW_ERR;
+    swoole_trace("fd=%d, events=%d", socket->fd, events);
+    reactor_->_set(socket, events);
+
+    return SW_OK;
 }
 
 int ReactorPoll::del(Socket *socket) {
@@ -144,62 +130,46 @@ int ReactorPoll::del(Socket *socket) {
         return SW_ERR;
     }
 
-    for (uint32_t i = 0; i < reactor_->get_event_num(); i++) {
-        if (events_[i].fd == socket->fd) {
-            for (; i < reactor_->get_event_num(); i++) {
-                if (i == reactor_->get_event_num() - 1) {
-                    fds_[i] = nullptr;
-                    events_[i].fd = 0;
-                    events_[i].events = 0;
-                } else {
-                    fds_[i] = fds_[i + 1];
-                    events_[i] = events_[i + 1];
-                }
-            }
-            reactor_->_del(socket);
-            return SW_OK;
-        }
+    if (!reactor_->_exists(socket)) {
+        swoole_error_log(SW_LOG_WARNING,
+                         SW_ERROR_SOCKET_NOT_EXISTS,
+                         "[Reactor#%d] failed to delete events[fd=%d, fd_type=%d], the socket#%d is not exists",
+                         reactor_->id,
+                         socket->fd,
+                         socket->fd_type,
+                         socket->fd);
+        swoole_print_backtrace_on_error();
+        return SW_ERR;
     }
 
-    swoole_error_log(SW_LOG_WARNING,
-                     SW_ERROR_SOCKET_NOT_EXISTS,
-                     "[Reactor#%d] failed to delete events[fd=%d, fd_type=%d], the socket#%d is not exists",
-                     reactor_->id,
-                     socket->fd,
-                     socket->fd_type,
-                     socket->fd);
-    swoole_print_backtrace_on_error();
-
-    return SW_ERR;
+    reactor_->_del(socket);
+    return SW_OK;
 }
 
 int ReactorPoll::wait() {
     Event event;
     ReactorHandler handler;
 
-    int ret;
     reactor_->before_wait();
 
     while (reactor_->running) {
         reactor_->execute_begin_callback();
-
-        ret = poll(events_, reactor_->get_event_num(), reactor_->get_timeout_msec());
+        const int event_num = set_events();
+        int ret = poll(events_, event_num, reactor_->get_timeout_msec());
         if (ret < 0) {
             if (!reactor_->catch_error()) {
-                swoole_sys_warning("[Reactor#%d] poll(nfds=%zu, timeout=%d) failed",
+                swoole_sys_warning("[Reactor#%d] poll(nfds=%d, timeout=%d) failed",
                                    reactor_->id,
-                                   reactor_->get_event_num(),
+                                   event_num,
                                    reactor_->get_timeout_msec());
                 break;
-            } else {
-                goto _continue;
             }
         } else if (ret == 0) {
             reactor_->execute_end_callbacks(true);
             SW_REACTOR_CONTINUE;
         } else {
-            for (uint32_t i = 0; i < reactor_->get_event_num(); i++) {
-                event.socket = fds_[i];
+            for (int i = 0; i < event_num; i++) {
+                event.socket = reactor_->get_socket(events_[i].fd);
                 event.fd = events_[i].fd;
                 event.reactor_id = reactor_->id;
                 event.type = event.socket->fd_type;
@@ -245,20 +215,39 @@ int ReactorPoll::wait() {
                 }
             }
         }
-    _continue:
         reactor_->execute_end_callbacks(false);
         SW_REACTOR_CONTINUE;
     }
     return SW_OK;
 }
 
-bool ReactorPoll::exists(int fd) const {
-    for (uint32_t i = 0; i < reactor_->get_event_num(); i++) {
-        if (events_[i].fd == fd) {
-            return true;
-        }
+int16_t translate_events_to_poll(int events) {
+    int16_t poll_events = 0;
+
+    if (events & SW_EVENT_READ) {
+        poll_events |= POLLIN;
+    }
+    if (events & SW_EVENT_WRITE) {
+        poll_events |= POLLOUT;
     }
-    return false;
+
+    return poll_events;
 }
 
+int translate_events_from_poll(int16_t events) {
+    int sw_events = 0;
+
+    if (events & POLLIN) {
+        sw_events |= SW_EVENT_READ;
+    }
+    if (events & POLLOUT) {
+        sw_events |= SW_EVENT_WRITE;
+    }
+    // ignore ERR and HUP, because event is already processed at IN and OUT handler.
+    if ((((events & POLLERR) || (events & POLLHUP)) && !((events & POLLIN) || (events & POLLOUT)))) {
+        sw_events |= SW_EVENT_ERROR;
+    }
+
+    return sw_events;
+}
 }  // namespace swoole