event.c

快速开发

/* 
 * The contents of this file are subject to the Mozilla Public
 * License Version 1.1 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of
 * the License at http://www.mozilla.org/MPL/
 * 
 * Software distributed under the License is distributed on an "AS
 * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * rights and limitations under the License.
 * 
 * The Original Code is the Netscape Portable Runtime library.
 * 
 * The Initial Developer of the Original Code is Netscape
 * Communications Corporation.  Portions created by Netscape are 
 * Copyright (C) 1994-2000 Netscape Communications Corporation.  All
 * Rights Reserved.
 * 
 * Contributor(s):  Silicon Graphics, Inc.
 *                  Yahoo! Inc.
 *
 * Alternatively, the contents of this file may be used under the
 * terms of the GNU General Public License Version 2 or later (the
 * "GPL"), in which case the provisions of the GPL are applicable 
 * instead of those above.  If you wish to allow use of your 
 * version of this file only under the terms of the GPL and not to
 * allow others to use your version of this file under the MPL,
 * indicate your decision by deleting the provisions above and
 * replace them with the notice and other provisions required by
 * the GPL.  If you do not delete the provisions above, a recipient
 * may use your version of this file under either the MPL or the
 * GPL.
 */

#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <time.h>
#include <errno.h>
#include "common.h"

#ifdef MD_HAVE_KQUEUE
#include <sys/event.h>
#endif
#ifdef MD_HAVE_EPOLL
#include <sys/epoll.h>
#endif

#if defined(USE_POLL) && !defined(MD_HAVE_POLL)
/* Force poll usage if explicitly asked for it */
#define MD_HAVE_POLL
#endif


static struct _st_seldata {
    fd_set fd_read_set, fd_write_set, fd_exception_set;
    int fd_ref_cnts[FD_SETSIZE][3];
    int maxfd;
} *_st_select_data;

#define _ST_SELECT_MAX_OSFD      (_st_select_data->maxfd)
#define _ST_SELECT_READ_SET      (_st_select_data->fd_read_set)
#define _ST_SELECT_WRITE_SET     (_st_select_data->fd_write_set)
#define _ST_SELECT_EXCEP_SET     (_st_select_data->fd_exception_set)
#define _ST_SELECT_READ_CNT(fd)  (_st_select_data->fd_ref_cnts[fd][0])
#define _ST_SELECT_WRITE_CNT(fd) (_st_select_data->fd_ref_cnts[fd][1])
#define _ST_SELECT_EXCEP_CNT(fd) (_st_select_data->fd_ref_cnts[fd][2])


#ifdef MD_HAVE_POLL
static struct _st_polldata {
    struct pollfd *pollfds;
    int pollfds_size;
    int fdcnt;
} *_st_poll_data;

#define _ST_POLL_OSFD_CNT        (_st_poll_data->fdcnt) 
#define _ST_POLLFDS              (_st_poll_data->pollfds) 
#define _ST_POLLFDS_SIZE         (_st_poll_data->pollfds_size)
#endif  /* MD_HAVE_POLL */


#ifdef MD_HAVE_KQUEUE
typedef struct _kq_fd_data {
    int rd_ref_cnt;
    int wr_ref_cnt;
    int revents;
} _kq_fd_data_t;

static struct _st_kqdata {
    _kq_fd_data_t *fd_data;
    struct kevent *evtlist;
    struct kevent *addlist;
    struct kevent *dellist;
    int fd_data_size;
    int evtlist_size;
    int addlist_size;
    int addlist_cnt;
    int dellist_size;
    int dellist_cnt;
    int kq;
    pid_t pid;
} *_st_kq_data;

#ifndef ST_KQ_MIN_EVTLIST_SIZE
#define ST_KQ_MIN_EVTLIST_SIZE 64
#endif

#define _ST_KQ_READ_CNT(fd)      (_st_kq_data->fd_data[fd].rd_ref_cnt)
#define _ST_KQ_WRITE_CNT(fd)     (_st_kq_data->fd_data[fd].wr_ref_cnt)
#define _ST_KQ_REVENTS(fd)       (_st_kq_data->fd_data[fd].revents)
#endif  /* MD_HAVE_KQUEUE */


#ifdef MD_HAVE_EPOLL
typedef struct _epoll_fd_data {
    int rd_ref_cnt;
    int wr_ref_cnt;
    int ex_ref_cnt;
    int revents;
} _epoll_fd_data_t;

static struct _st_epolldata {
    _epoll_fd_data_t *fd_data;
    struct epoll_event *evtlist;
    int fd_data_size;
    int evtlist_size;
    int evtlist_cnt;
    int fd_hint;
    int epfd;
    pid_t pid;
} *_st_epoll_data;

#ifndef ST_EPOLL_EVTLIST_SIZE
/* Not a limit, just a hint */
#define ST_EPOLL_EVTLIST_SIZE 4096
#endif

#define _ST_EPOLL_READ_CNT(fd)   (_st_epoll_data->fd_data[fd].rd_ref_cnt)
#define _ST_EPOLL_WRITE_CNT(fd)  (_st_epoll_data->fd_data[fd].wr_ref_cnt)
#define _ST_EPOLL_EXCEP_CNT(fd)  (_st_epoll_data->fd_data[fd].ex_ref_cnt)
#define _ST_EPOLL_REVENTS(fd)    (_st_epoll_data->fd_data[fd].revents)

#define _ST_EPOLL_READ_BIT(fd)   (_ST_EPOLL_READ_CNT(fd) ? EPOLLIN : 0)
#define _ST_EPOLL_WRITE_BIT(fd)  (_ST_EPOLL_WRITE_CNT(fd) ? EPOLLOUT : 0)
#define _ST_EPOLL_EXCEP_BIT(fd)  (_ST_EPOLL_EXCEP_CNT(fd) ? EPOLLPRI : 0)
#define _ST_EPOLL_EVENTS(fd) \
    (_ST_EPOLL_READ_BIT(fd)|_ST_EPOLL_WRITE_BIT(fd)|_ST_EPOLL_EXCEP_BIT(fd))

#endif  /* MD_HAVE_EPOLL */

_st_eventsys_t *_st_eventsys = NULL;


/*****************************************
 * select event system
 */

int _st_select_init(void)
{
    _st_select_data = (struct _st_seldata *) malloc(sizeof(*_st_select_data));
    if (!_st_select_data)
        return -1;

    memset(_st_select_data, 0, sizeof(*_st_select_data));
    _st_select_data->maxfd = -1;

    return 0;
}

int _st_select_pollset_add(struct pollfd *pds, int npds)
{
    struct pollfd *pd;
    struct pollfd *epd = pds + npds;

    /* Do checks up front */
    for (pd = pds; pd < epd; pd++) {
        if (pd->fd < 0 || pd->fd >= FD_SETSIZE || !pd->events ||
            (pd->events & ~(POLLIN | POLLOUT | POLLPRI))) {
            errno = EINVAL;
            return -1;
        }
    }

    for (pd = pds; pd < epd; pd++) {
        if (pd->events & POLLIN) {
            FD_SET(pd->fd, &_ST_SELECT_READ_SET);
            _ST_SELECT_READ_CNT(pd->fd)++;
        }
        if (pd->events & POLLOUT) {
            FD_SET(pd->fd, &_ST_SELECT_WRITE_SET);
            _ST_SELECT_WRITE_CNT(pd->fd)++;
        }
        if (pd->events & POLLPRI) {
            FD_SET(pd->fd, &_ST_SELECT_EXCEP_SET);
            _ST_SELECT_EXCEP_CNT(pd->fd)++;
        }
        if (_ST_SELECT_MAX_OSFD < pd->fd)
            _ST_SELECT_MAX_OSFD = pd->fd;
    }

    return 0;
}

void _st_select_pollset_del(struct pollfd *pds, int npds)
{
    struct pollfd *pd;
    struct pollfd *epd = pds + npds;

    for (pd = pds; pd < epd; pd++) {
        if (pd->events & POLLIN) {
            if (--_ST_SELECT_READ_CNT(pd->fd) == 0)
                FD_CLR(pd->fd, &_ST_SELECT_READ_SET);
        }
        if (pd->events & POLLOUT) {
            if (--_ST_SELECT_WRITE_CNT(pd->fd) == 0)
                FD_CLR(pd->fd, &_ST_SELECT_WRITE_SET);
        }
        if (pd->events & POLLPRI) {
            if (--_ST_SELECT_EXCEP_CNT(pd->fd) == 0)
                FD_CLR(pd->fd, &_ST_SELECT_EXCEP_SET);
        }
    }
}

void _st_select_find_bad_fd(void)
{
    _st_clist_t *q;
    _st_pollq_t *pq;
    int notify;
    struct pollfd *pds, *epds;
    int pq_max_osfd, osfd;
    short events;

    _ST_SELECT_MAX_OSFD = -1;

    for (q = _ST_IOQ.next; q != &_ST_IOQ; q = q->next) {
        pq = _ST_POLLQUEUE_PTR(q);
        notify = 0;
        epds = pq->pds + pq->npds;
        pq_max_osfd = -1;
      
        for (pds = pq->pds; pds < epds; pds++) {
            osfd = pds->fd;
            pds->revents = 0;
            if (pds->events == 0)
                continue;
            if (fcntl(osfd, F_GETFL, 0) < 0) {
                pds->revents = POLLNVAL;
                notify = 1;
            }
            if (osfd > pq_max_osfd) {
                pq_max_osfd = osfd;
            }
        }

        if (notify) {
            ST_REMOVE_LINK(&pq->links);
            pq->on_ioq = 0;
            /*
             * Decrement the count of descriptors for each descriptor/event
             * because this I/O request is being removed from the ioq
             */
            for (pds = pq->pds; pds < epds; pds++) {
                osfd = pds->fd;
                events = pds->events;
                if (events & POLLIN) {
                    if (--_ST_SELECT_READ_CNT(osfd) == 0) {
                        FD_CLR(osfd, &_ST_SELECT_READ_SET);
                    }
                }
                if (events & POLLOUT) {
                    if (--_ST_SELECT_WRITE_CNT(osfd) == 0) {
                        FD_CLR(osfd, &_ST_SELECT_WRITE_SET);
                    }
                }
                if (events & POLLPRI) {
                    if (--_ST_SELECT_EXCEP_CNT(osfd) == 0) {
                        FD_CLR(osfd, &_ST_SELECT_EXCEP_SET);
                    }
                }
            }

            if (pq->thread->flags & _ST_FL_ON_SLEEPQ)
                _ST_DEL_SLEEPQ(pq->thread);
            pq->thread->state = _ST_ST_RUNNABLE;
            _ST_ADD_RUNQ(pq->thread);
        } else {
            if (_ST_SELECT_MAX_OSFD < pq_max_osfd)
                _ST_SELECT_MAX_OSFD = pq_max_osfd;
        }
    }
}

void _st_select_dispatch(void)
{
    struct timeval timeout, *tvp;
    fd_set r, w, e;
    fd_set *rp, *wp, *ep;
    int nfd, pq_max_osfd, osfd;
    _st_clist_t *q;
    st_utime_t min_timeout;
    _st_pollq_t *pq;
    int notify;
    struct pollfd *pds, *epds;
    short events, revents;

    /*
     * Assignment of fd_sets
     */
    r = _ST_SELECT_READ_SET;
    w = _ST_SELECT_WRITE_SET;
    e = _ST_SELECT_EXCEP_SET;

    rp = &r;
    wp = &w;
    ep = &e;

    if (_ST_SLEEPQ == NULL) {
        tvp = NULL;
    } else {
        min_timeout = (_ST_SLEEPQ->due <= _ST_LAST_CLOCK) ? 0 :
            (_ST_SLEEPQ->due - _ST_LAST_CLOCK);
        timeout.tv_sec  = (int) (min_timeout / 1000000);
        timeout.tv_usec = (int) (min_timeout % 1000000);
        tvp = &timeout;
    }

    /* Check for I/O operations */
    nfd = select(_ST_SELECT_MAX_OSFD + 1, rp, wp, ep, tvp);

    /* Notify threads that are associated with the selected descriptors */
    if (nfd > 0) {
        _ST_SELECT_MAX_OSFD = -1;
        for (q = _ST_IOQ.next; q != &_ST_IOQ; q = q->next) {
            pq = _ST_POLLQUEUE_PTR(q);
            notify = 0;
            epds = pq->pds + pq->npds;
            pq_max_osfd = -1;
      
            for (pds = pq->pds; pds < epds; pds++) {
                osfd = pds->fd;
                events = pds->events;
                revents = 0;
                if ((events & POLLIN) && FD_ISSET(osfd, rp)) {
                    revents |= POLLIN;
                }
                if ((events & POLLOUT) && FD_ISSET(osfd, wp)) {
                    revents |= POLLOUT;
                }
                if ((events & POLLPRI) && FD_ISSET(osfd, ep)) {
                    revents |= POLLPRI;
                }
                pds->revents = revents;
                if (revents) {
                    notify = 1;
                }
                if (osfd > pq_max_osfd) {
                    pq_max_osfd = osfd;
                }
            }
            if (notify) {
                ST_REMOVE_LINK(&pq->links);
                pq->on_ioq = 0;
                /*
                 * Decrement the count of descriptors for each descriptor/event
                 * because this I/O request is being removed from the ioq
                 */
                for (pds = pq->pds; pds < epds; pds++) {
                    osfd = pds->fd;
                    events = pds->events;
                    if (events & POLLIN) {
                        if (--_ST_SELECT_READ_CNT(osfd) == 0) {
                            FD_CLR(osfd, &_ST_SELECT_READ_SET);
                        }
                    }
                    if (events & POLLOUT) {
                        if (--_ST_SELECT_WRITE_CNT(osfd) == 0) {
                            FD_CLR(osfd, &_ST_SELECT_WRITE_SET);
                        }
                    }
                    if (events & POLLPRI) {
                        if (--_ST_SELECT_EXCEP_CNT(osfd) == 0) {
                            FD_CLR(osfd, &_ST_SELECT_EXCEP_SET);
                        }
                    }
                }

                if (pq->thread->flags & _ST_FL_ON_SLEEPQ)
                    _ST_DEL_SLEEPQ(pq->thread);
                pq->thread->state = _ST_ST_RUNNABLE;
                _ST_ADD_RUNQ(pq->thread);
            } else {
                if (_ST_SELECT_MAX_OSFD < pq_max_osfd)
                    _ST_SELECT_MAX_OSFD = pq_max_osfd;
            }
        }
    } else if (nfd < 0) {
        /*
         * It can happen when a thread closes file descriptor
         * that is being used by some other thread -- BAD!
         */
        if (errno == EBADF)
            _st_select_find_bad_fd();
    }
}

int _st_select_fd_new(int osfd)
{
    if (osfd >= FD_SETSIZE) {
        errno = EMFILE;
        return -1;
    }

    return 0;
}

int _st_select_fd_close(int osfd)
{
    if (_ST_SELECT_READ_CNT(osfd) || _ST_SELECT_WRITE_CNT(osfd) ||
        _ST_SELECT_EXCEP_CNT(osfd)) {
        errno = EBUSY;
        return -1;
    }

    return 0;
}

int _st_select_fd_getlimit(void)
{
    return FD_SETSIZE;
}

static _st_eventsys_t _st_select_eventsys = {
    "select",
    ST_EVENTSYS_SELECT,
    _st_select_init,
    _st_select_dispatch,
    _st_select_pollset_add,
    _st_select_pollset_del,
    _st_select_fd_new,
    _st_select_fd_close,
    _st_select_fd_getlimit
};


#ifdef MD_HAVE_POLL
/*****************************************
 * poll event system
 */

int _st_poll_init(void)
{
    _st_poll_data = (struct _st_polldata *) malloc(sizeof(*_st_poll_data));
    if (!_st_poll_data)
        return -1;

    _ST_POLLFDS = (struct pollfd *) malloc(ST_MIN_POLLFDS_SIZE *
                                           sizeof(struct pollfd));
    if (!_ST_POLLFDS) {
        free(_st_poll_data);
        _st_poll_data = NULL;
        return -1;
    }
    _ST_POLLFDS_SIZE = ST_MIN_POLLFDS_SIZE;
    _ST_POLL_OSFD_CNT = 0;

    return 0;
}

int _st_poll_pollset_add(struct pollfd *pds, int npds)
{
    struct pollfd *pd;
    struct pollfd *epd = pds + npds;

    for (pd = pds; pd < epd; pd++) {
        if (pd->fd < 0 || !pd->events) {
            errno = EINVAL;
            return -1;
        }
    }