fdset.c File Reference

#include "gw-config.h"
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include "gwlib/gwlib.h"

Go to the source code of this file.

Data Structures
struct	FDSet
struct	action

Functions
static struct action *	action_create (int type)
static void	action_destroy (struct action *action)
static void	action_destroy_item (void *action)
static void	submit_action (FDSet *set, struct action *action)
static void	submit_action_nosync (FDSet *set, struct action *action)
static int	handle_action (FDSet *set, struct action *action)
static int	find_entry (FDSet *set, int fd)
static void	remove_entry (FDSet *set, int entry)
static void	remove_deleted_entries (FDSet *set)
static void	poller (void *arg)
FDSet *	fdset_create_real (long timeout)
void	fdset_destroy (FDSet *set)
void	fdset_register (FDSet *set, int fd, int events, fdset_callback_t callback, void *data)
void	fdset_listen (FDSet *set, int fd, int mask, int events)
void	fdset_unregister (FDSet *set, int fd)
void	fdset_set_timeout (FDSet *set, long timeout)

Function Documentation

action_create()

static struct action* action_create ( int  type )

static

Definition at line 155 of file fdset.c.

References type.

Referenced by fdset_destroy(), fdset_listen(), fdset_register(), fdset_set_timeout(), and fdset_unregister().

{
    struct action *new;

    new = gw_malloc(sizeof(*new));
    new->type = type;
    new->fd = -1;
    new->mask = 0;
    new->events = 0;
    new->callback = NULL;
    new->data = NULL;
    new->done = NULL;

    return new;
}

action_destroy()

static void action_destroy ( struct action *  action )

static

Definition at line 171 of file fdset.c.

References action::done, and gwlist_destroy().

Referenced by action_destroy_item(), handle_action(), and submit_action().

{
    if (action == NULL)
        return;

    gwlist_destroy(action->done, NULL);
    gw_free(action);
}

action_destroy_item()

static void action_destroy_item ( void *  action )

static

Definition at line 181 of file fdset.c.

References action_destroy().

Referenced by fdset_destroy().

{
    action_destroy(action);
}

fdset_create_real()

FDSet* fdset_create_real

(

long

timeout

)

Definition at line 368 of file fdset.c.

References error(), fdset_destroy(), gwlist_create, gwthread_create, poller(), and action::timeout.

Referenced by port_add(), and start_client_threads().

{
    FDSet *new;

    new = gw_malloc(sizeof(*new));

    /* Start off with space for one element because we can't malloc 0 bytes
     * and we don't want to worry about these pointers being NULL. */
    new->size = 1;
    new->entries = 0;
    new->pollinfo = gw_malloc(sizeof(new->pollinfo[0]) * new->size);
    new->callbacks = gw_malloc(sizeof(new->callbacks[0]) * new->size);
    new->datafields = gw_malloc(sizeof(new->datafields[0]) * new->size);
    new->times = gw_malloc(sizeof(new->times[0]) * new->size);
    new->timeout = timeout > 0 ? timeout : -1;
    new->scanning = 0;
    new->deleted_entries = 0;

    new->actions = gwlist_create();

    new->poll_thread = gwthread_create(poller, new);
    if (new->poll_thread < 0) {
        error(0, "Could not start internal thread for fdset.");
        fdset_destroy(new);
        return NULL;
    }

    return new;
}

fdset_destroy()

void fdset_destroy

(

FDSet *

set

)

Definition at line 398 of file fdset.c.

References action_create(), action_destroy_item(), action::DESTROY, error(), gwlist_destroy(), gwlist_len(), gwthread_join(), gwthread_self(), submit_action(), and warning().

Referenced by client_shutdown(), fdset_create_real(), handle_action(), port_remove(), and start_client_threads().

{
    if (set == NULL)
        return;

    if (set->poll_thread < 0 || gwthread_self() == set->poll_thread) {
        if (set->entries > 0) {
            warning(0, "Destroying fdset with %d active entries.",
                    set->entries);
        }
        gw_free(set->pollinfo);
        gw_free(set->callbacks);
        gw_free(set->datafields);
        gw_free(set->times);
        if (gwlist_len(set->actions) > 0) {
            error(0, "Destroying fdset with %ld pending actions.",
                  gwlist_len(set->actions));
        }
        gwlist_destroy(set->actions, action_destroy_item);
        gw_free(set);
    } else {
        long thread = set->poll_thread;
        submit_action(set, action_create(DESTROY));
    gwthread_join(thread);
    }
}

fdset_listen()

void fdset_listen	(	FDSet *	set,
		int	fd,
		int	mask,
		int	events
	)

Definition at line 470 of file fdset.c.

References action(), action_create(), action::events, action::fd, find_entry(), gw_assert(), gwthread_self(), action::LISTEN, action::mask, submit_action(), and warning().

Referenced by handle_action(), unlocked_register_pollin(), and unlocked_register_pollout().

{
    int entry;

    gw_assert(set != NULL);

    if (gwthread_self() != set->poll_thread) {
        struct action *action;

        action = action_create(LISTEN);
        action->fd = fd;
    action->mask = mask;
        action->events = events;
        submit_action(set, action);
        return;
    }

    entry = find_entry(set, fd);   
    if (entry < 0) {
        warning(0, "fdset_listen called on unregistered fd %d.", fd);
        return;
    }

    /* Copy the bits from events specified by the mask, and preserve the
     * bits not specified by the mask. */
    set->pollinfo[entry].events =
    (set->pollinfo[entry].events & ~mask) | (events & mask);

    /* If poller is currently scanning the array, then change the
     * revents field so that the callback function will not be called
     * for events we should no longer listen for.  The idea is the
     * same as for the events field, except that we only turn bits off. */
    if (set->scanning) {
        set->pollinfo[entry].revents =
            set->pollinfo[entry].revents & (events | ~mask);
    }
    
    time(&set->times[entry]);
}

fdset_register()

void fdset_register	(	FDSet *	set,
		int	fd,
		int	events,
		fdset_callback_t	callback,
		void *	data
	)

Definition at line 425 of file fdset.c.

References action(), action_create(), action::callback, action::data, action::events, action::fd, gw_assert(), gwthread_self(), action::REGISTER, and submit_action_nosync().

Referenced by conn_register_real(), and handle_action().

{
    int new;

    gw_assert(set != NULL);

    if (gwthread_self() != set->poll_thread) {
        struct action *action;

        action = action_create(REGISTER);
        action->fd = fd;
        action->events = events;
        action->callback = callback;
        action->data = data;
    submit_action_nosync(set, action);
        return;
    }

    gw_assert(set->entries <= set->size);

    if (set->entries >= set->size) {
        int newsize = set->entries + 1;
        set->pollinfo = gw_realloc(set->pollinfo,
                                   sizeof(set->pollinfo[0]) * newsize);
        set->callbacks = gw_realloc(set->callbacks,
                                   sizeof(set->callbacks[0]) * newsize);
        set->datafields = gw_realloc(set->datafields,
                                   sizeof(set->datafields[0]) * newsize);
        set->times = gw_realloc(set->times, sizeof(set->times[0]) * newsize);
        set->size = newsize;
    }

    /* We don't check set->scanning.  Adding new entries is not harmful
     * because their revents fields are 0. */

    new = set->entries++;
    set->pollinfo[new].fd = fd;
    set->pollinfo[new].events = events;
    set->pollinfo[new].revents = 0;
    set->callbacks[new] = callback;
    set->datafields[new] = data;
    time(&set->times[new]);
}

fdset_set_timeout()

void fdset_set_timeout	(	FDSet *	set,
		long	timeout
	)

Set timeout in seconds for this FDSet.

Definition at line 547 of file fdset.c.

References action(), action_create(), gw_assert(), gwthread_self(), action::SET_TIMEOUT, submit_action(), and action::timeout.

Referenced by http_set_client_timeout(), and port_set_timeout().

{
    gw_assert(set != NULL);

    if (gwthread_self() != set->poll_thread) {
        struct action *action;

        action = action_create(SET_TIMEOUT);
        action->timeout = timeout;
        submit_action(set, action);
        return;
    }
    set->timeout = timeout;
}

fdset_unregister()

void fdset_unregister	(	FDSet *	set,
		int	fd
	)

Definition at line 510 of file fdset.c.

References action(), action_create(), action::fd, find_entry(), gw_assert(), gwthread_self(), remove_entry(), submit_action(), action::UNREGISTER, and warning().

Referenced by conn_destroy(), conn_unregister(), and handle_action().

{
    int entry;

    gw_assert(set != NULL);

    if (gwthread_self() != set->poll_thread) {
        struct action *action;

        action = action_create(UNREGISTER);
        action->fd = fd;
        submit_action(set, action);
        return;
    }

    /* Remove the entry from the pollinfo array */

    entry = find_entry(set, fd);
    if (entry < 0) {
        warning(0, "fdset_listen called on unregistered fd %d.", fd);
        return;
    }

    if (entry == set->entries - 1) {
        /* It's the last entry.  We can safely remove it even while
         * the array is being scanned, because the scan checks set->entries. */
        set->entries--;
    } else if (set->scanning) {
        /* We can't remove entries because the array is being
         * scanned.  Mark it as deleted.  */
        set->pollinfo[entry].fd = -1;
        set->deleted_entries++;
    } else {
        remove_entry(set, entry);
    }
}

find_entry()

static int find_entry ( FDSet *  set, int  fd  )

static

Definition at line 269 of file fdset.c.

References action::fd, gw_assert(), and gwthread_self().

Referenced by fdset_listen(), and fdset_unregister().

{
    int i;

    gw_assert(set != NULL);
    gw_assert(gwthread_self() == set->poll_thread);

    for (i = 0; i < set->entries; i++) {
        if (set->pollinfo[i].fd == fd)
            return i;
    }

    return -1;
}

handle_action()

static int handle_action ( FDSet *  set, struct action *  action  )

static

Definition at line 226 of file fdset.c.

References action_destroy(), action::callback, action::data, action::DESTROY, action::done, action::events, action::fd, fdset_destroy(), fdset_listen(), fdset_register(), fdset_unregister(), gw_assert(), gwlist_produce(), gwthread_self(), action::LISTEN, action::mask, panic, action::REGISTER, action::SET_TIMEOUT, action::timeout, action::type, and action::UNREGISTER.

Referenced by poller().

{
    int result;

    gw_assert(set != NULL);
    gw_assert(set->poll_thread == gwthread_self());
    gw_assert(action != NULL);

    result = 0;

    switch (action->type) {
    case REGISTER:
        fdset_register(set, action->fd, action->events,
                       action->callback, action->data);
        break;
    case LISTEN:
        fdset_listen(set, action->fd, action->mask, action->events);
        break;
    case UNREGISTER:
        fdset_unregister(set, action->fd);
        break;
    case DESTROY:
        fdset_destroy(set);
        result = -1;
        break;
    case SET_TIMEOUT:
        set->timeout = action->timeout;
        break;
    default:
        panic(0, "fdset: handle_action got unknown action type %d.",
              action->type);
    }

    if (action->done == NULL)
    action_destroy(action);
    else
        gwlist_produce(action->done, action);

    return result;
}

poller()

static void poller ( void *  arg )

static

Definition at line 318 of file fdset.c.

References action(), debug(), error(), gw_assert(), gwlist_extract_first(), gwthread_poll(), gwthread_sleep(), handle_action(), POLLERR, and remove_deleted_entries().

Referenced by fdset_create_real().

{
    FDSet *set = arg;
    struct action *action;
    int ret;
    int i;
    time_t now;

    gw_assert(set != NULL);

    for (;;) {
        while ((action = gwlist_extract_first(set->actions)) != NULL) {
            /* handle_action returns -1 if the set was destroyed. */
            if (handle_action(set, action) < 0)
                return;
        }

        /* Block for defined timeout, waiting for activity */
        ret = gwthread_poll(set->pollinfo, set->entries, set->timeout);

        if (ret < 0) {
            if (errno != EINTR) {
                error(errno, "Poller: can't handle error; sleeping 1 second.");
                gwthread_sleep(1.0);
            }
            continue;
        }
        time(&now);
        /* Callbacks may modify the table while we scan it, so be careful. */
        set->scanning = 1;
        for (i = 0; i < set->entries; i++) {
            if (set->pollinfo[i].revents != 0) {
                set->callbacks[i](set->pollinfo[i].fd,
                                set->pollinfo[i].revents,
                                set->datafields[i]);
                /* update event time */
                time(&set->times[i]);
            } else if (set->timeout > 0 && difftime(set->times[i] + set->timeout, now) <= 0) {
                debug("gwlib.fdset", 0, "Timeout for fd:%d appears.", set->pollinfo[i].fd);
                set->callbacks[i](set->pollinfo[i].fd, POLLERR, set->datafields[i]);
            }
        }
        set->scanning = 0;

    if (set->deleted_entries > 0)
        remove_deleted_entries(set);
    }
}

remove_deleted_entries()

static void remove_deleted_entries ( FDSet *  set )

static

Definition at line 297 of file fdset.c.

References remove_entry().

Referenced by poller().

{
    int i;

    i = 0;
    while (i < set->entries && set->deleted_entries > 0) {
        if (set->pollinfo[i].fd < 0) {
            remove_entry(set, i);
        set->deleted_entries--;
    } else {
        i++;
        }
    }
}

remove_entry()

static void remove_entry ( FDSet *  set, int  entry  )

static

Definition at line 284 of file fdset.c.

Referenced by fdset_unregister(), and remove_deleted_entries().

{
    if (entry != set->entries - 1) {
        /* We need to keep the array contiguous, so move the last element
         * to fill in the hole. */
        set->pollinfo[entry] = set->pollinfo[set->entries - 1];
        set->callbacks[entry] = set->callbacks[set->entries - 1];
        set->datafields[entry] = set->datafields[set->entries - 1];
        set->times[entry] = set->times[set->entries - 1];
    }
    set->entries--;
}

submit_action()

static void submit_action ( FDSet *  set, struct action *  action  )

static

Definition at line 191 of file fdset.c.

References action_destroy(), action::done, gw_assert(), gwlist_add_producer(), gwlist_append(), gwlist_consume(), gwlist_create, and gwthread_wakeup().

Referenced by fdset_destroy(), fdset_listen(), fdset_set_timeout(), and fdset_unregister().

{
    List *done;
    void *sync;

    gw_assert(set != NULL);
    gw_assert(action != NULL);

    done = gwlist_create();
    gwlist_add_producer(done);

    action->done = done;

    gwlist_append(set->actions, action);
    gwthread_wakeup(set->poll_thread);

    sync = gwlist_consume(done);
    gw_assert(sync == action);

    action_destroy(action);
}

submit_action_nosync()

static void submit_action_nosync ( FDSet *  set, struct action *  action  )

static

Definition at line 216 of file fdset.c.

References gwlist_append(), and gwthread_wakeup().

Referenced by fdset_register().

{
    gwlist_append(set->actions, action);
    gwthread_wakeup(set->poll_thread);
}