text: /*
   +----------------------------------------------------------------------+
   | Zan                                                                  |
   +----------------------------------------------------------------------+
   | Copyright (c) 2012-2016 Swoole Team <http://github.com/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       |
   | license@swoole.com so we can mail you a copy immediately.            |
   +----------------------------------------------------------------------+
   | Author: Tianfeng Han  <mikan.tenny@gmail.com>                        |
   +----------------------------------------------------------------------+
   */
  
  
  #include "swoole.h"
  #include "swError.h"
  #include "swReactor.h"
  
  #include "zanLog.h"
  
  #ifdef HAVE_EPOLL
  
  #include <sys/epoll.h>
  
  #ifndef EPOLLRDHUP
  #define EPOLLRDHUP   0x2000
  #define NO_EPOLLRDHUP
  #endif
  
  #ifndef EPOLLONESHOT
  #define EPOLLONESHOT (1u << 30)
  #endif
  
  typedef struct swReactorEpoll_s swReactorEpoll;
  
  typedef struct _swFd
  {
      uint32_t fd;
      uint32_t fdtype;
  } swFd;
  
  static int swReactorEpoll_add(swReactor *reactor, int fd, int fdtype);
  static int swReactorEpoll_set(swReactor *reactor, int fd, int fdtype);
  static int swReactorEpoll_del(swReactor *reactor, int fd);
  static int swReactorEpoll_wait(swReactor *reactor, struct timeval *timeo);
  static void swReactorEpoll_free(swReactor *reactor);
  
  static sw_inline int swReactorEpoll_event_set(int fdtype)
  {
      uint32_t flag = 0;
      if (swReactor_event_read(fdtype))
      {
          flag |= EPOLLIN;
      }
      if (swReactor_event_write(fdtype))
      {
          flag |= EPOLLOUT;
      }
      if (swReactor_event_error(fdtype))
      {
          flag |= (EPOLLRDHUP | EPOLLHUP | EPOLLERR);
      }
  
      return flag;
  }
  
  struct swReactorEpoll_s
  {
      int epfd;
      struct epoll_event *events;
  };
  
  int swReactorEpoll_create(swReactor *reactor, int max_event_num)
  {
      //create reactor object
      swReactorEpoll *reactor_object = sw_malloc(sizeof(swReactorEpoll));
      if (reactor_object == NULL)
      {
          zanWarn("malloc[0] failed.");
          return ZAN_ERR;
      }
      bzero(reactor_object, sizeof(swReactorEpoll));
      reactor->object = reactor_object;
      reactor->max_event_num = max_event_num;
  
      reactor_object->events = sw_calloc(max_event_num, sizeof(struct epoll_event));
  
      if (reactor_object->events == NULL)
      {
          zanWarn("malloc[1] failed.");
          sw_free(reactor_object);
          return ZAN_ERR;
      }
      //epoll create
      reactor_object->epfd = epoll_create(512);
      if (reactor_object->epfd < 0)
      {
          zanWarn("epoll_create failed.");
          sw_free(reactor_object->events);
          sw_free(reactor_object);
          return ZAN_ERR;
      }
  
      //binding method
      reactor->add = swReactorEpoll_add;
      reactor->set = swReactorEpoll_set;
      reactor->del = swReactorEpoll_del;
      reactor->wait = swReactorEpoll_wait;
      reactor->free = swReactorEpoll_free;
  
      return ZAN_OK;
  }
  
  static void swReactorEpoll_free(swReactor *reactor)
  {
      swReactorEpoll *object = reactor->object;
      close(object->epfd);
      sw_free(object->events);
      sw_free(object);
  }
  
  static int swReactorEpoll_add(swReactor *reactor, int fd, int fdtype)
  {
      if (swReactor_add(reactor, fd, fdtype) < 0)
      {
          return ZAN_ERR;
      }
  
      swReactorEpoll *object = reactor->object;
      struct epoll_event e;
      swFd fd_;
      int ret;
      bzero(&e, sizeof(struct epoll_event));
  
      fd_.fd = fd;
      fd_.fdtype = swReactor_fdtype(fdtype);
      e.events = swReactorEpoll_event_set(fdtype);
  
      memcpy(&(e.data.u64), &fd_, sizeof(fd_));
      ret = epoll_ctl(object->epfd, EPOLL_CTL_ADD, fd, &e);
      if (ret < 0)
      {
          zanWarn("add events[fd=%d#%d, type=%d, events=%d] failed.", fd, reactor->id, fd_.fdtype, e.events);
          return ZAN_ERR;
      }
  
      zanTrace("add event[reactor_id=%d|fd=%d]", reactor->id, fd);
      reactor->event_num++;
      return ZAN_OK;
  }
  
  static int swReactorEpoll_del(swReactor *reactor, int fd)
  {
      swReactorEpoll *object = reactor->object;
      int ret;
  
      if (fd < 0)
      {
          return ZAN_ERR;
      }
      ret = epoll_ctl(object->epfd, EPOLL_CTL_DEL, fd, NULL);
      if (ret < 0)
      {
          zanWarn("epoll remove fd[%d#%d] failed.", fd, reactor->id);
          return ZAN_ERR;
      }
  
      if (swReactor_del(reactor, fd) < 0)
      {
          return ZAN_ERR;
      }
  
      reactor->event_num = reactor->event_num <= 0 ? 0 : reactor->event_num - 1;
      zanTrace("remove event[reactor_id=%d|fd=%d]", reactor->id, fd);
      return ZAN_OK;
  }
  
  static int swReactorEpoll_set(swReactor *reactor, int fd, int fdtype)
  {
      swReactorEpoll *object = reactor->object;
      swFd fd_;
      struct epoll_event e;
      int ret;
  
      bzero(&e, sizeof(struct epoll_event));
      e.events = swReactorEpoll_event_set(fdtype);
  
      if (e.events & EPOLLOUT)
      {
          assert(fd > 2);
      }
  
      fd_.fd = fd;
      fd_.fdtype = swReactor_fdtype(fdtype);
      memcpy(&(e.data.u64), &fd_, sizeof(fd_));
  
      ret = epoll_ctl(object->epfd, EPOLL_CTL_MOD, fd, &e);
      if (ret < 0)
      {
          zanWarn("reactor#%d->set(fd=%d|type=%d|events=%d) failed.", reactor->id, fd, fd_.fdtype, e.events);
          return ZAN_ERR;
      }
      //execute parent method
      swReactor_set(reactor, fd, fdtype);
      return ZAN_OK;
  }
  
  static int swReactorEpoll_wait(swReactor *reactor, struct timeval *timeo)
  {
      swEvent event;
      swReactorEpoll *object = reactor->object;
      swReactor_handle handle;
      int i, n, ret, msec;
  
      int reactor_id = reactor->id;
      int epoll_fd = object->epfd;
      int max_event_num = reactor->max_event_num;
      struct epoll_event *events = object->events;
  
      if (reactor->timeout_msec == 0)
      {
              reactor->timeout_msec = (timeo == NULL)? -1:timeo->tv_sec * 1000 + timeo->tv_usec / 1000;
      }
  
      while (reactor->running > 0)
      {
          msec = reactor->timeout_msec;
          /// 等待超时时间矫正，防止有defer 需要执行，但epoll_wait没有唤醒
  //        msec = msec < 0 && reactor->defer_callback_list? 1:msec;
          msec = reactor->defer_callback_list? 1:msec;
          n = epoll_wait(epoll_fd, events, max_event_num, msec);
          if (n < 0)
          {
              if (swReactor_error(reactor) < 0)
              {
                  zanWarn("[Reactor#%d] epoll_wait failed.", reactor_id);
                  return ZAN_ERR;
              }
              else
              {
                  continue;
              }
          }
          else if (n == 0)
          {
              if (reactor->onTimeout != NULL)
              {
                  reactor->onTimeout(reactor);
              }
          }
  
          for (i = 0; i < n; i++)
          {
              event.fd = events[i].data.u64;
              event.from_id = reactor_id;
              event.type = events[i].data.u64 >> 32;
              event.socket = swReactor_get(reactor, event.fd);
              if (event.socket->fd != event.fd) {
                  zanDebug("EPOLL event.socket->fd[%d] != event.fd[%d]",event.socket->fd,event.fd);
              }
  
              event.socket->event_trigger = 1;
              //error
  #ifndef NO_EPOLLRDHUP
              if ((events[i].events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) &&
                      !event.socket->removed && event.socket->event_trigger)
  #else
              if ((events[i].events & (EPOLLERR | EPOLLHUP)) &&
                      !event.socket->removed && event.socket->event_trigger)
  #endif
              {
                  handle = swReactor_getHandle(reactor, SW_EVENT_ERROR, event.type);
                  if (handle)
                  {
                      ret = handle(reactor, &event);
                      if (ret < 0)
                      {
                          zanWarn("EPOLLERR handle failed. fd=%d.", event.fd);
                      }
                  }
  
                  event.socket->event_trigger = 0;
                  continue;
              }
  
              //read
              if ((events[i].events & EPOLLIN) &&
                      !event.socket->removed && event.socket->event_trigger)
              {
                  handle = swReactor_getHandle(reactor, SW_EVENT_READ, event.type);
                  if (handle)
                  {
                      ret =  handle(reactor, &event);
                      if (ret < 0)
                      {
                          zanWarn("EPOLLIN handle failed. fd=%d.", event.fd);
                      }
                  }
              }
              //write
              if ((events[i].events & EPOLLOUT) &&
                      !event.socket->removed && event.socket->event_trigger)
              {
                  handle = swReactor_getHandle(reactor, SW_EVENT_WRITE, event.type);
                  if (handle)
                  {
                      ret = handle(reactor, &event);
                      if (ret < 0)
                      {
                          zanWarn("EPOLLOUT handle failed. fd=%d.", event.fd);
                      }
                  }
              }
  
              event.socket->event_trigger = 0;
          }
  
          if (reactor->onFinish != NULL)
          {
              reactor->onFinish(reactor);
          }
      }
      return 0;
  }
  
  #endif
  