/*
 +----------------------------------------------------------------------+
 | 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
