event

event.h

#ifndef _EVENT_H_
#define _EVENT_H_

#define EVLIST_TIMEOUT    0x01
#define EVLIST_READ    0x02
#define EVLIST_WRITE    0x04
#define EVLIST_ADD    0x08
#define EVLIST_INIT    0x80

#define EV_TIMEOUT    EVLIST_TIMEOUT
#define EV_READ        EVLIST_READ
#define EV_WRITE    EVLIST_WRITE

struct event {
    struct {
        struct event *tqe_next;    /* next element */
        struct event **tqe_prev;    /* address of previous next element */
    }ev_read_next;

    struct {
        struct event *tqe_next;    /* next element */
        struct event **tqe_prev;    /* address of previous next element */
    }ev_write_next;

    struct {
        struct event *tqe_next;    /* next element */
        struct event **tqe_prev;    /* address of previous next element */
    }ev_timeout_next;

    struct {
        struct event *tqe_next;    /* next element */
        struct event **tqe_prev;    /* address of previous next element */
    }ev_add_next;

    int ev_fd;
    short ev_events;

    struct timeval ev_timeout;

    void (*ev_callback)(int, short, void *arg);
    void *ev_arg;

    int ev_flags;
};

#define TIMEOUT_DEFAULT    5

void event_init(void);
int event_dispatch(void);

int timeout_next(struct timeval *);
void timeout_process(void);

#define timeout_add(ev, tv)        event_add(ev, tv)
#define timeout_set(ev, cb, arg)    event_set(ev, -1, 0, cb, arg)
#define timeout_del(ev)            event_del(ev)
#define timeout_pending(ev, tv)        event_pending(ev, EV_TIMEOUT, tv)
#define timeout_initalized(ev)        ((ev)->ev_flags & EVLIST_INIT)

void event_set(struct event *, int, short, void (*)(int, short, void *), void *);
void event_add(struct event *, struct timeval *);
void event_del(struct event *);

int event_pending(struct event *, short, struct timeval *);

#define event_initalized(ev)        ((ev)->ev_flags & EVLIST_INIT)

#endif /* _EVENT_H_ */

event.c

#include <sys/types.h>
#include <sys/time.h>
#include <sys/queue.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>

#ifdef USE_LOG
#include "log.h"
#else
#define LOG_DBG(x)
#define log_error(x)    perror(x)
#endif

#include "event.h"

#ifndef howmany
#define        howmany(x, y) (((x)+((y)-1))/(y))
#endif

/* Prototypes */
void event_add_post(struct event *);

struct timeout_list {
    struct event *tqh_first;    /* first element */
    struct event **tqh_last;    /* addr of last next element */
}timequeue;

struct event_wlist {
    struct event *tqh_first;    /* first element */
    struct event **tqh_last;    /* addr of last next element */
}writequeue;

struct event_rlist {
    struct event *tqh_first;    /* first element */
    struct event **tqh_last;    /* addr of last next element */
}readqueue;

struct event_ilist {
    struct event *tqh_first;    /* first element */
    struct event **tqh_last;    /* addr of last next element */
}addqueue;

int event_inloop = 0;
int event_fds;        /* Highest fd in fd set */
int event_fdsz;
fd_set *event_readset;
fd_set *event_writeset;

void
event_init(void)
{
    (&timequeue)->tqh_first = NULL;
    (&timequeue)->tqh_last = &(&timequeue)->tqh_first;

    (&writequeue)->tqh_first = NULL;
    (&writequeue)->tqh_last = &(&writequeue)->tqh_first;

    (&readqueue)->tqh_first = NULL;
    (&readqueue)->tqh_last = &(&readqueue)->tqh_first;

    (&addqueue)->tqh_first = NULL;
    (&addqueue)->tqh_last = &(&addqueue)->tqh_first;;
}

/*
 * Called with the highest fd that we know about. If it is 0, completely
 * recalculate everything.
 */

int
events_recalc(int max)
{
    fd_set *readset, *writeset;
    struct event *ev;
    int fdsz;

    event_fds = max;

    if (!event_fds) {

    //TAILQ_FOREACH(ev, &writequeue, ev_write_next)
        for(ev = TAILQ_FIRST(&writequeue);
                ev != TAILQ_END(&writequeue);
                ev = TAILQ_NEXT(ev,ev_write_next))
        {
            if (ev->ev_fd > event_fds)
                event_fds = ev->ev_fd;
        }
        for(ev = TAILQ_FIRST(&readqueue);
              ev != TAILQ_END(&readqueue);
              ev = TAILQ_NEXT(ev, ev_read_next))
        {
            if (ev->ev_fd > event_fds)
                event_fds = ev->ev_fd;
        }
    }

    fdsz = howmany(event_fds + 1, NFDBITS) * sizeof(fd_mask);
    if (fdsz > event_fdsz) {
        if ((readset = realloc(event_readset, fdsz)) == NULL) {
            log_error("malloc");
            return (-1);
        }

        if ((writeset = realloc(event_writeset, fdsz)) == NULL) {
            log_error("malloc");
            free(readset);
            return (-1);
        }

        memset(readset + event_fdsz, 0, fdsz - event_fdsz);
        memset(writeset + event_fdsz, 0, fdsz - event_fdsz);

        event_readset = readset;
        event_writeset = writeset;
        event_fdsz = fdsz;
    }

    return (0);
}

int
event_dispatch(void)
{
    struct timeval tv;
    struct event *ev, *old;
    int res, maxfd;

    /* Calculate the initial events that we are waiting for */
    if (events_recalc(0) == -1)
        return (-1);

    while (1) {
        memset(event_readset, 0, event_fdsz);
        memset(event_writeset, 0, event_fdsz);
//        TAILQ_FOREACH(var, head, field)
//    for((var) = TAILQ_FIRST(head);(var) != TAILQ_END(head);(var) = TAILQ_NEXT(var, field))
        for(ev = TAILQ_FIRST(&writequeue);
                ev != TAILQ_END(&writequeue);
                ev = TAILQ_NEXT(ev, ev_write_next))
        {
                FD_SET(ev->ev_fd, event_writeset);
        }
        for(ev = TAILQ_FIRST(&readqueue);
                ev != TAILQ_END(&readqueue);
                ev = TAILQ_NEXT(ev, ev_read_next))
        {
                FD_SET(ev->ev_fd, event_readset);
        }

        timeout_next(&tv);

        if ((res = select(event_fds + 1, event_readset,
                  event_writeset, NULL, &tv)) == -1) {
            if (errno != EINTR) {
                log_error("select");
                return (-1);
            }
            continue;
        }

        LOG_DBG((LOG_MISC, 80, __FUNCTION__": select reports %d",
             res));

        maxfd = 0;
        event_inloop = 1;
        for (ev = TAILQ_FIRST(&readqueue); ev;) {
            old = TAILQ_NEXT(ev, ev_read_next);
            if (FD_ISSET(ev->ev_fd, event_readset)) {
                event_del(ev);
                (*ev->ev_callback)(ev->ev_fd, EV_READ,
                           ev->ev_arg);
            } else if (ev->ev_fd > maxfd)
                maxfd = ev->ev_fd;

            ev = old;
        }

        for (ev = TAILQ_FIRST(&writequeue); ev;) {
            old = TAILQ_NEXT(ev, ev_read_next);
            if (FD_ISSET(ev->ev_fd, event_writeset)) {
                event_del(ev);
                (*ev->ev_callback)(ev->ev_fd, EV_WRITE,
                           ev->ev_arg);
            } else if (ev->ev_fd > maxfd)
                maxfd = ev->ev_fd;

            ev = old;
        }
        event_inloop = 0;

        for (ev = TAILQ_FIRST(&addqueue); ev;
             ev = TAILQ_FIRST(&addqueue)) {
            TAILQ_REMOVE(&addqueue, ev, ev_add_next);
            ev->ev_flags &= ~EVLIST_ADD;

            event_add_post(ev);

            if (ev->ev_fd > maxfd)
                maxfd = ev->ev_fd;
        }

        if (events_recalc(maxfd) == -1)
            return (-1);

        timeout_process();
    }

    return (0);
}

void
event_set(struct event *ev, int fd, short events,
      void (*callback)(int, short, void *), void *arg)
{
    ev->ev_callback = callback;
    ev->ev_arg = arg;
    ev->ev_fd = fd;
    ev->ev_events = events;
    ev->ev_flags = EVLIST_INIT;
}

/*
 * Checks if a specific event is pending or scheduled.
 */

int
event_pending(struct event *ev, short event, struct timeval *tv)
{
    int flags = ev->ev_flags;

    /*
     * We might not have been able to add it to the actual queue yet,
     * check if we will enqueue later.
     */
    if (ev->ev_flags & EVLIST_ADD)
        flags |= (ev->ev_events & (EV_READ|EV_WRITE));

    event &= (EV_TIMEOUT|EV_READ|EV_WRITE);

    /* See if there is a timeout that we should report */
    if (tv != NULL && (flags & event & EV_TIMEOUT))
        *tv = ev->ev_timeout;

    return (flags & event);
}

void
event_add(struct event *ev, struct timeval *tv)
{
    LOG_DBG((LOG_MISC, 55,
         "event_add: event: %p, %s%s%scall %p",
         ev,
         ev->ev_events & EV_READ ? "EV_READ " : " ",
         ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
         tv ? "EV_TIMEOUT " : " ",
         ev->ev_callback));
    if (tv != NULL) {
        struct timeval now;
        struct event *tmp;

        gettimeofday(&now, NULL);
        timeradd(&now, tv, &ev->ev_timeout);

        LOG_DBG((LOG_MISC, 55,
             "event_add: timeout in %d seconds, call %p",
             tv->tv_sec, ev->ev_callback));
        if (ev->ev_flags & EVLIST_TIMEOUT)
            TAILQ_REMOVE(&timequeue, ev, ev_timeout_next);

        /* Insert in right temporal order */
        for (tmp = TAILQ_FIRST(&timequeue); tmp;
             tmp = TAILQ_NEXT(tmp, ev_timeout_next)) {
         if (timercmp(&ev->ev_timeout, &tmp->ev_timeout, <=))
                 break;
        }

        if (tmp)
            TAILQ_INSERT_BEFORE(tmp, ev, ev_timeout_next);
        else
            TAILQ_INSERT_TAIL(&timequeue, ev, ev_timeout_next);

        ev->ev_flags |= EVLIST_TIMEOUT;
    }

    if (event_inloop) {
        /* We are in the event loop right now, we have to
         * postpone the change until later.
         */
        if (ev->ev_flags & EVLIST_ADD)
            return;

        TAILQ_INSERT_TAIL(&addqueue, ev, ev_add_next);
        ev->ev_flags |= EVLIST_ADD;
    } else
        event_add_post(ev);
}

void
event_add_post(struct event *ev)
{
    if ((ev->ev_events & EV_READ) && !(ev->ev_flags & EVLIST_READ)) {
        TAILQ_INSERT_TAIL(&readqueue, ev, ev_read_next);

        ev->ev_flags |= EVLIST_READ;
    }

    if ((ev->ev_events & EV_WRITE) && !(ev->ev_flags & EVLIST_WRITE)) {
        TAILQ_INSERT_TAIL(&writequeue, ev, ev_write_next);

        ev->ev_flags |= EVLIST_WRITE;
    }
}

void
event_del(struct event *ev)
{
    LOG_DBG((LOG_MISC, 80, "event_del: %p, callback %p",
         ev, ev->ev_callback));

    if (ev->ev_flags & EVLIST_ADD) {
        TAILQ_REMOVE(&addqueue, ev, ev_add_next);

        ev->ev_flags &= ~EVLIST_ADD;
    }

    if (ev->ev_flags & EVLIST_TIMEOUT) {
        TAILQ_REMOVE(&timequeue, ev, ev_timeout_next);

        ev->ev_flags &= ~EVLIST_TIMEOUT;
    }

    if (ev->ev_flags & EVLIST_READ) {
        TAILQ_REMOVE(&readqueue, ev, ev_read_next);

        ev->ev_flags &= ~EVLIST_READ;
    }

    if (ev->ev_flags & EVLIST_WRITE) {
        TAILQ_REMOVE(&writequeue, ev, ev_write_next);

        ev->ev_flags &= ~EVLIST_WRITE;
    }
}

int
timeout_next(struct timeval *tv)
{
    struct timeval now;
    struct event *ev;

    if ((ev = TAILQ_FIRST(&timequeue)) == NULL) {
        timerclear(tv);
        tv->tv_sec = TIMEOUT_DEFAULT;
        return (0);
    }

    if (gettimeofday(&now, NULL) == -1)
        return (-1);

    if (timercmp(&ev->ev_timeout, &now, <=)) {
        timerclear(tv);
        return (0);
    }

    timersub(&ev->ev_timeout, &now, tv);

    LOG_DBG((LOG_MISC, 60, "timeout_next: in %d seconds", tv->tv_sec));
    return (0);
}

void
timeout_process(void)
{
    struct timeval now;
    struct event *ev;

    gettimeofday(&now, NULL);

    while ((ev = TAILQ_FIRST(&timequeue)) != NULL) {
        if (timercmp(&ev->ev_timeout, &now, >))
            break;

        TAILQ_REMOVE(&timequeue, ev, ev_timeout_next);
        ev->ev_flags &= ~EVLIST_TIMEOUT;

        LOG_DBG((LOG_MISC, 60, "timeout_process: call %p",
             ev->ev_callback));
        (*ev->ev_callback)(ev->ev_fd, EV_TIMEOUT, ev->ev_arg);
    }
}

event-test.c

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/queue.h>
#include <sys/time.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>

#include <event.h>

void
fifo_read(int fd, short event, void *arg)
{
    char buf[255];
    int len;
    struct event *ev = arg;

    /* Reschedule this event */
    event_add(ev, NULL);

    fprintf(stderr, "fifo_read called with fd: %d, event: %d, arg: %p\n",
        fd, event, arg);

    len = read(fd, buf, sizeof(buf) - 1);
    if (len == -1) {
        perror("read");
        return;
    } else if (len == 0) {
        fprintf(stderr, "Connection closed\n");
        return;
    }

    buf[len] = '\0';
    fprintf(stdout, "Read: %s\n", buf);
}

int
main (int argc, char **argv)
{
    struct stat st;
    char *fifo = "event.fifo";
    int socket;
    struct event evfifo;

    if (lstat (fifo, &st) == 0) {
        if ((st.st_mode & S_IFMT) == S_IFREG) {
            errno = EEXIST;
            perror("lstat");
            exit (1);
        }
    }

    unlink (fifo);
    if (mkfifo (fifo, 0600) == -1) {
        perror("mkfifo");
        exit (1);
    }

    /* Linux pipes are broken, we need O_RDWR instead of O_RDONLY */
#ifdef __linux
    socket = open (fifo, O_RDWR | O_NONBLOCK, 0);
#else
    socket = open (fifo, O_RDONLY | O_NONBLOCK, 0);
#endif

    if (socket == -1) {
        perror("open");
        exit (1);
    }

    fprintf(stderr, "Write data to %s\n", fifo);

    /* Initalize the event library */
    event_init();

    /* Initalize one event */
    event_set(&evfifo, socket, EV_READ, fifo_read, &evfifo);

    /* Add it to the active events, without a timeout */
    event_add(&evfifo, NULL);

    event_dispatch();

    return (0);
}