Ensure event cache handle properly (#202)

src/Poller/EventPoller.cpp

@@ -124,6 +124,7 @@ int EventPoller::addEvent(int fd, int event, PollEventCB cb) {
         }
 #endif
         auto record = std::make_shared<Poll_Record>();
+        record->fd = fd;
         record->event = event;
         record->call_back = std::move(cb);
         _event_map.emplace(fd, record);
@@ -146,10 +147,12 @@ int EventPoller::delEvent(int fd, PollCompleteCB cb) {
     if (isCurrentThread()) {
 #if defined(HAS_EPOLL)
         bool success = epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == 0 && _event_map.erase(fd) > 0;
+        refreshEventCache(fd);
         cb(success);
         return success ? 0 : -1;
 #else
         cb(_event_map.erase(fd));
+        refreshEventCache(fd);
         return 0;
 #endif //HAS_EPOLL
 
@@ -303,9 +306,17 @@ void EventPoller::runLoop(bool blocked, bool ref_self) {
                 //超时或被打断
                 continue;
             }
+
+            _event_cache_expired_map.clear();
+
             for (int i = 0; i < ret; ++i) {
                 struct epoll_event &ev = events[i];
                 int fd = ev.data.fd;
+                if (_event_cache_expired_map[fd]) {
+                    //event cache refresh
+                    continue;
+                }
+
                 auto it = _event_map.find(fd);
                 if (it == _event_map.end()) {
                     epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
@@ -357,6 +368,9 @@ void EventPoller::runLoop(bool blocked, bool ref_self) {
                 //超时或被打断
                 continue;
             }
+
+            _event_cache_expired_map.clear();
+
             //收集select事件类型
             for (auto &pr : _event_map) {
                 int event = 0;
@@ -375,7 +389,12 @@ void EventPoller::runLoop(bool blocked, bool ref_self) {
                 }
             }
 
-            callback_list.for_each([](Poll_Record::Ptr &record) {
+            callback_list.for_each([this](Poll_Record::Ptr &record) {
+                if (this->_event_cache_expired_map[record->fd]) {
+                    //event cache refresh
+                    return;
+                }
+
                 try {
                     record->call_back(record->attach);
                 } catch (std::exception &ex) {
@@ -445,6 +464,11 @@ EventPoller::DelayTask::Ptr EventPoller::doDelayTask(uint64_t delay_ms, function
     return ret;
 }
 
+void EventPoller::refreshEventCache(int fd) {
+    _event_cache_expired_map[fd] = true;
+    return;
+}
+
 ///////////////////////////////////////////////
 
 static size_t s_pool_size = 0;

src/Poller/EventPoller.h

@@ -178,6 +178,11 @@ class EventPoller : public TaskExecutor, public AnyStorage, public std::enable_s
      */
     void addEventPipe();
 
+    /**
+     * 删除过时的事件缓存
+     */
+    void refreshEventCache(int fd);
+
 private:
     class ExitException : public std::exception {};
 
@@ -210,12 +215,14 @@ class EventPoller : public TaskExecutor, public AnyStorage, public std::enable_s
     //select相关
     struct Poll_Record {
         using Ptr = std::shared_ptr<Poll_Record>;
+        int fd;
         int event;
         int attach;
         PollEventCB call_back;
     };
     unordered_map<int, Poll_Record::Ptr> _event_map;
 #endif //HAS_EPOLL
+    unordered_map<int, bool> _event_cache_expired_map;
 
     //定时器相关
     std::multimap<uint64_t, DelayTask::Ptr> _delay_task_map;
@@ -277,4 +284,4 @@ class EventPollerPool : public std::enable_shared_from_this<EventPollerPool>, pu
 };
 
 }  // namespace toolkit
-#endif /* EventPoller_h */
+#endif /* EventPoller_h */