PHP FPM源代碼反芻品味之六：IO事件機制的封裝.

服務器程序繞不過對操做系統IO事件機制(多路複用)的使用,
不一樣的操做系統，支持不一樣的IO事件機制，linux 支持epoll,
windows支持select, freebsd 支持kqueue. 詳細以下:

select     (任何 POSIX os, 含windows)
poll       (任何 POSIX os)
epoll      (linux >= 2.5.44)
kqueue     (FreeBSD >= 4.1, OpenBSD >= 2.9, NetBSD >= 2.0)
/dev/poll  (Solaris >= 7)
port       (Solaris >= 10)

程序要支持跨平臺,須要支持不一樣的機制,但不一樣的機制有不一樣的使用接口(api),
爲了省事,須要有個統一的接口來處理. 這個需求早就有聰明人爲咱們考慮到了.
現成的庫libevent, libev, libuv就是對不一樣操做系統IO事件機制的封裝,
要開發服務程序,選一個就可用.

PHP FPM 沒有使用上述封裝好的庫, 而是本身實現了一個簡單的封裝.
咱們看看FPM怎麼封裝的? 同時也瞭解下C語言對函數的封裝實現.

FPM IO事件機制封裝的特色.

1. IO事件的文件句柄fd 和 事件處理函數(回調函數) 封裝到一個對象中(c結構體)
2. 設計一個新的對象實現不一樣函數的封裝.
3. 編譯時選擇事件機制, 編譯時經過對操做系統的檢測,選擇支持的機制.

源代碼註釋說明

事件對象結構

//fpm_event.h
struct fpm_event_s {
    int fd;                   /* IO 文件句柄*/
    struct timeval timeout;   
    struct timeval frequency;
    void (*callback)(struct fpm_event_s *, short, void *); /* 回調函數 */
    void *arg; /* 回調函數的參數 */
    int flags;
    int index;         
    short which;       
};
//隊列,多個事件對象的容器
typedef struct fpm_event_queue_s {
    struct fpm_event_queue_s *prev;
    struct fpm_event_queue_s *next;
    struct fpm_event_s *ev;
} fpm_event_queue;

事件模塊封裝結構

struct fpm_event_module_s {
    const char *name;
    int support_edge_trigger;
    int (*init)(int max_fd);
    int (*clean)(void);
    //等待多個事件
    int (*wait)(struct fpm_event_queue_s *queue, unsigned long int timeout);
    int (*add)(struct fpm_event_s *ev);
    int (*remove)(struct fpm_event_s *ev);
};

這個結構很清晰:
就是對事件對象的 添加,移除,等待操做,
再加上使用先後的初始化及清理操做.

對事件的添加,移除,等待也是不一樣實現機制的同一的一個接口.
因爲事件回調函數及其參數也放到了事件對象裏, 等到事件發生時,觸發回調函數便可.

瞭解的外觀接口後,咱們看看不一樣機制模塊的實現, 以epoll,select爲例.

epoll

//events/epoll.c
//環境變量,在編譯時肯定是否歸入.
#if HAVE_EPOLL

#include <sys/epoll.h>
#include <errno.h>

static int fpm_event_epoll_init(int max);
static int fpm_event_epoll_clean();
static int fpm_event_epoll_wait(struct fpm_event_queue_s *queue, unsigned long int timeout);
static int fpm_event_epoll_add(struct fpm_event_s *ev);
static int fpm_event_epoll_remove(struct fpm_event_s *ev);

static struct fpm_event_module_s epoll_module = {
    .name = "epoll",
    .support_edge_trigger = 1,
    .init = fpm_event_epoll_init,
    .clean = fpm_event_epoll_clean,
    .wait = fpm_event_epoll_wait,
    .add = fpm_event_epoll_add,
    .remove = fpm_event_epoll_remove, 
};

//全局變量
static struct epoll_event *epollfds = NULL;
static int nepollfds = 0;
static int epollfd = -1;

#endif /* HAVE_EPOLL */

//這是使用時,獲取模塊對象的函數
//系統不支持返回NULL,這個函數老是編入二進制文件裏.
struct fpm_event_module_s *fpm_event_epoll_module() /* {{{ */
{
#if HAVE_EPOLL
    return &epoll_module;
#else
    return NULL;
#endif /* HAVE_EPOLL */
}
/* }}} */

#if HAVE_EPOLL

/*
 * Init the module
 */
static int fpm_event_epoll_init(int max) /* {{{ */
{
    if (max < 1) {
        return 0;
    }

    /* init epoll */
    epollfd = epoll_create(max + 1);
    if (epollfd < 0) {
        zlog(ZLOG_ERROR, "epoll: unable to initialize");
        return -1;
    }

    /* allocate fds */
    epollfds = malloc(sizeof(struct epoll_event) * max);
    if (!epollfds) {
        zlog(ZLOG_ERROR, "epoll: unable to allocate %d events", max);
        return -1;
    }
    memset(epollfds, 0, sizeof(struct epoll_event) * max);

    /* save max */
    nepollfds = max;

    return 0;
}
/* }}} */

/*
 * Clean the module
 */
static int fpm_event_epoll_clean() /* {{{ */
{
    /* free epollfds */
    if (epollfds) {
        free(epollfds);
        epollfds = NULL;
    }
    if (epollfd != -1) {
        close(epollfd);
        epollfd = -1;
    }

    nepollfds = 0;

    return 0;
}
/* }}} */

/*
 * wait for events or timeout
 */
static int fpm_event_epoll_wait(struct fpm_event_queue_s *queue, unsigned long int timeout) /* {{{ */
{
    int ret, i;

    /* ensure we have a clean epoolfds before calling epoll_wait() */
    memset(epollfds, 0, sizeof(struct epoll_event) * nepollfds);

    /* wait for inconming event or timeout */
    ret = epoll_wait(epollfd, epollfds, nepollfds, timeout);
    if (ret == -1) {

        /* trigger error unless signal interrupt */
        if (errno != EINTR) {
            zlog(ZLOG_WARNING, "epoll_wait() returns %d", errno);
            return -1;
        }
    }

    /* events have been triggered, let's fire them */
    for (i = 0; i < ret; i++) {

        /* do we have a valid ev ptr ? */
        if (!epollfds[i].data.ptr) {
            continue;
        }

        /* fire the event */
        fpm_event_fire((struct fpm_event_s *)epollfds[i].data.ptr);

        /* sanity check */
        if (fpm_globals.parent_pid != getpid()) {
            return -2;
        }
    }

    return ret;
}
/* }}} */

/*
 * Add a FD to the fd set
 */
static int fpm_event_epoll_add(struct fpm_event_s *ev) /* {{{ */
{
    struct epoll_event e;

    /* fill epoll struct */
    e.events = EPOLLIN;
    e.data.fd = ev->fd;

    //data.ptr 設爲自定義對象, 事件觸發時,以此獲取自定義對象
    e.data.ptr = (void *)ev;

    if (ev->flags & FPM_EV_EDGE) {
        e.events = e.events | EPOLLET;
    }

    /* add the event to epoll internal queue */
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, ev->fd, &e) == -1) {
        zlog(ZLOG_ERROR, "epoll: unable to add fd %d", ev->fd);
        return -1;
    }

    /* mark the event as registered */
    ev->index = ev->fd;
    return 0;
}
/* }}} */

/*
 * Remove a FD from the fd set
 */
static int fpm_event_epoll_remove(struct fpm_event_s *ev) /* {{{ */
{
    struct epoll_event e;

    /* fill epoll struct the same way we did in fpm_event_epoll_add() */
    e.events = EPOLLIN;
    e.data.fd = ev->fd;
    e.data.ptr = (void *)ev;

    if (ev->flags & FPM_EV_EDGE) {
        e.events = e.events | EPOLLET;
    }

    /* remove the event from epoll internal queue */
    if (epoll_ctl(epollfd, EPOLL_CTL_DEL, ev->fd, &e) == -1) {
        zlog(ZLOG_ERROR, "epoll: unable to remove fd %d", ev->fd);
        return -1;
    }

    /* mark the event as not registered */
    ev->index = -1;
    return 0;
}
/* }}} */

#endif /* HAVE_EPOLL */

select

////events/select.c
//環境變量,在編譯時肯定是否歸入.
#if HAVE_SELECT

/* According to POSIX.1-2001 */
#include <sys/select.h>

/* According to earlier standards */
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include <errno.h>

static int fpm_event_select_init(int max);
static int fpm_event_select_wait(struct fpm_event_queue_s *queue, unsigned long int timeout);
static int fpm_event_select_add(struct fpm_event_s *ev);
static int fpm_event_select_remove(struct fpm_event_s *ev);

static struct fpm_event_module_s select_module = {
    .name = "select",
    .support_edge_trigger = 0,
    .init = fpm_event_select_init,
    .clean = NULL,
    .wait = fpm_event_select_wait,
    .add = fpm_event_select_add,
    .remove = fpm_event_select_remove,
};

//全局變量
static fd_set fds;

#endif /* HAVE_SELECT */

//這是使用時,獲取模塊對象的函數
//系統不支持返回NULL,這個函數老是編入二進制文件裏.
struct fpm_event_module_s *fpm_event_select_module() /* {{{ */
{
#if HAVE_SELECT
    return &select_module;
#else
    return NULL;
#endif /* HAVE_SELECT */
}
/* }}} */

#if HAVE_SELECT

/*
 * Init the module
 */
static int fpm_event_select_init(int max) /* {{{ */
{
    FD_ZERO(&fds);
    return 0;
}
/* }}} */


/*
 * wait for events or timeout
 */
static int fpm_event_select_wait(struct fpm_event_queue_s *queue, unsigned long int timeout) /* {{{ */
{
    int ret;
    struct fpm_event_queue_s *q;
    fd_set current_fds;
    struct timeval t;

    /* copy fds because select() alters it */
    current_fds = fds;

    /* fill struct timeval with timeout */
    t.tv_sec = timeout / 1000;
    t.tv_usec = (timeout % 1000) * 1000;

    /* wait for inconming event or timeout */
    ret = select(FD_SETSIZE, &current_fds, NULL, NULL, &t);
    if (ret == -1) {

        /* trigger error unless signal interrupt */
        if (errno != EINTR) {
            zlog(ZLOG_WARNING, "poll() returns %d", errno);
            return -1;
        }
    }

    /* events have been triggered */
    if (ret > 0) {

        /* trigger POLLIN events */
        q = queue;
        while (q) {
            if (q->ev) { /* sanity check */

                /* check if the event has been triggered */
                if (FD_ISSET(q->ev->fd, &current_fds)) {

                    /* fire the event */
                    fpm_event_fire(q->ev);

                    /* sanity check */
                    if (fpm_globals.parent_pid != getpid()) {
                        return -2;
                    }
                }
            }
            q = q->next; /* iterate */
        }
    }
    return ret;

}
/* }}} */

/*
 * Add a FD to the fd set
 */
static int fpm_event_select_add(struct fpm_event_s *ev) /* {{{ */
{
    /* check size limitation */
    if (ev->fd >= FD_SETSIZE) {
        zlog(ZLOG_ERROR, "select: not enough space in the select fd list (max = %d). Please consider using another event mechanism.", FD_SETSIZE);
        return -1;
    }

    /* add the FD if not already in */
    if (!FD_ISSET(ev->fd, &fds)) {
        FD_SET(ev->fd, &fds);
        ev->index = ev->fd;
    }

    return 0;
}
/* }}} */

/*
 * Remove a FD from the fd set
 */
static int fpm_event_select_remove(struct fpm_event_s *ev) /* {{{ */
{
    /* remove the fd if it's in */
    if (FD_ISSET(ev->fd, &fds)) {
        FD_CLR(ev->fd, &fds);
        ev->index = -1;
    }

    return 0;
}
/* }}} */

#endif /* HAVE_SELECT */