text: /*
    +----------------------------------------------------------------------+
    | Open Swoole                                                          |
    +----------------------------------------------------------------------+
    | This source file is subject to version 2.0 of the Apache license,    |
    | that is bundled with this package in the file LICENSE, and is        |
    | available through the world-wide-web at the following url:           |
    | http://www.apache.org/licenses/LICENSE-2.0.html                      |
    | If you did not receive a copy of the Apache2.0 license and are unable|
    | to obtain it through the world-wide-web, please send a note to       |
    | hello@swoole.co.uk so we can mail you a copy immediately.            |
    +----------------------------------------------------------------------+
    | Author: Tianfeng Han  <mikan.tenny@gmail.com>                        |
    +----------------------------------------------------------------------+
  */
  
  #include <poll.h>
  
  #include "swoole.h"
  #include "swoole_socket.h"
  #include "swoole_reactor.h"
  
  namespace swoole {
  
  using network::Socket;
  
  class ReactorPoll : public ReactorImpl {
      uint32_t max_fd_num;
      Socket **fds_;
      struct pollfd *events_;
      bool exists(int fd);
  
    public:
      ReactorPoll(Reactor *_reactor, int max_events);
      ~ReactorPoll();
      bool ready() override {
          return true;
      };
      int add(Socket *socket, int events) override;
      int set(Socket *socket, int events) override;
      int del(Socket *socket) override;
      int wait(struct timeval *) override;
  };
  
  ReactorImpl *make_reactor_poll(Reactor *_reactor, int max_events) {
      return new ReactorPoll(_reactor, max_events);
  }
  
  ReactorPoll::ReactorPoll(Reactor *_reactor, int max_events) : ReactorImpl(_reactor) {
      fds_ = new Socket *[max_events];
      events_ = new struct pollfd[max_events];
  
      max_fd_num = max_events;
      reactor_->max_event_num = max_events;
  }
  
  ReactorPoll::~ReactorPoll() {
      delete[] fds_;
      delete[] events_;
  }
  
  int ReactorPoll::add(Socket *socket, int events) {
      int fd = socket->fd;
      if (exists(fd)) {
          swoole_warning("fd#%d is already exists", fd);
          return SW_ERR;
      }
  
      int cur = reactor_->event_num;
      if (reactor_->event_num == max_fd_num) {
          swoole_warning("too many connection, more than %d", max_fd_num);
          return SW_ERR;
      }
  
      reactor_->_add(socket, events);
  
      swoole_trace("fd=%d, events=%d", fd, events);
  
      fds_[cur] = socket;
      events_[cur].fd = fd;
      events_[cur].events = 0;
  
      if (Reactor::isset_read_event(events)) {
          events_[cur].events |= POLLIN;
      }
      if (Reactor::isset_write_event(events)) {
          events_[cur].events |= POLLOUT;
      }
      if (Reactor::isset_error_event(events)) {
          events_[cur].events |= POLLHUP;
      }
  
      return SW_OK;
  }
  
  int ReactorPoll::set(Socket *socket, int events) {
      uint32_t i;
  
      swoole_trace("fd=%d, events=%d", socket->fd, events);
  
      for (i = 0; i < reactor_->event_num; i++) {
          // found
          if (events_[i].fd == socket->fd) {
              events_[i].events = 0;
              if (Reactor::isset_read_event(events)) {
                  events_[i].events |= POLLIN;
              }
              if (Reactor::isset_write_event(events)) {
                  events_[i].events |= POLLOUT;
              }
              // execute parent method
              reactor_->_set(socket, events);
              return SW_OK;
          }
      }
  
      return SW_ERR;
  }
  
  int ReactorPoll::del(Socket *socket) {
      if (socket->removed) {
          swoole_error_log(SW_LOG_WARNING,
                           SW_ERROR_EVENT_SOCKET_REMOVED,
                           "failed to delete event[%d], it has already been removed",
                           socket->fd);
          return SW_ERR;
      }
  
      for (uint32_t i = 0; i < reactor_->event_num; i++) {
          if (events_[i].fd == socket->fd) {
              for (; i < reactor_->event_num; i++) {
                  if (i == reactor_->event_num) {
                      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;
          }
      }
  
      return SW_ERR;
  }
  
  int ReactorPoll::wait(struct timeval *timeo) {
      Event event;
      ReactorHandler handler;
  
      int ret;
  
      if (reactor_->timeout_msec == 0) {
          if (timeo == nullptr) {
              reactor_->timeout_msec = -1;
          } else {
              reactor_->timeout_msec = timeo->tv_sec * 1000 + timeo->tv_usec / 1000;
          }
      }
  
      reactor_->before_wait();
  
      while (reactor_->running) {
          if (reactor_->onBegin != nullptr) {
              reactor_->onBegin(reactor_);
          }
          ret = poll(events_, reactor_->event_num, reactor_->get_timeout_msec());
          if (ret < 0) {
              if (!reactor_->catch_error()) {
                  swoole_sys_warning("poll error");
                  break;
              } else {
                  goto _continue;
              }
          } else if (ret == 0) {
              reactor_->execute_end_callbacks(true);
              SW_REACTOR_CONTINUE;
          } else {
              for (uint32_t i = 0; i < reactor_->event_num; i++) {
                  event.socket = fds_[i];
                  event.fd = events_[i].fd;
                  event.reactor_id = reactor_->id;
                  event.type = event.socket->fd_type;
  
                  if (events_[i].revents & (POLLHUP | POLLERR)) {
                      event.socket->event_hup = 1;
                  }
  
                  swoole_trace("Event: fd=%d|reactor_id=%d|type=%d", event.fd, reactor_->id, event.type);
                  // in
                  if ((events_[i].revents & POLLIN) && !event.socket->removed) {
                      handler = reactor_->get_handler(SW_EVENT_READ, event.type);
                      ret = handler(reactor_, &event);
                      if (ret < 0) {
                          swoole_sys_warning("poll[POLLIN] handler failed. fd=%d", event.fd);
                      }
                  }
                  // out
                  if ((events_[i].revents & POLLOUT) && !event.socket->removed) {
                      handler = reactor_->get_handler(SW_EVENT_WRITE, event.type);
                      ret = handler(reactor_, &event);
                      if (ret < 0) {
                          swoole_sys_warning("poll[POLLOUT] handler failed. fd=%d", event.fd);
                      }
                  }
                  // error
                  if ((events_[i].revents & (POLLHUP | POLLERR)) && !event.socket->removed) {
                      // ignore ERR and HUP, because event is already processed at IN and OUT handler.
                      if ((events_[i].revents & POLLIN) || (events_[i].revents & POLLOUT)) {
                          continue;
                      }
                      handler = reactor_->get_error_handler(event.type);
                      ret = handler(reactor_, &event);
                      if (ret < 0) {
                          swoole_sys_warning("poll[POLLERR] handler failed. fd=%d", event.fd);
                      }
                  }
                  if (!event.socket->removed && (event.socket->events & SW_EVENT_ONCE)) {
                      del(event.socket);
                  }
              }
          }
      _continue:
          reactor_->execute_end_callbacks(false);
          SW_REACTOR_CONTINUE;
      }
      return SW_OK;
  }
  
  bool ReactorPoll::exists(int fd) {
      for (uint32_t i = 0; i < reactor_->event_num; i++) {
          if (events_[i].fd == fd) {
              return true;
          }
      }
      return false;
  }
  
  }  // namespace swoole
  