fdset.c

The Kannel Open Source WAP and SMS gateway works as both an SMS gateway, for implementing keyword b

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

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

    return -1;
}

static void remove_entry(FDSet *set, int entry)
{
    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->entries--;
}

static void remove_deleted_entries(FDSet *set)
{
    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++;
        }
    }
}

/* Main function for polling thread.  Most its time is spent blocking
 * in poll().  No-one else is allowed to change the fields it uses,
 * so other threads just put something on the actions list and wake
 * up this thread.  That's why it checks the actions list every time
 * it goes through the loop.
 */
static void poller(void *arg)
{
    FDSet *set = arg;
    struct action *action;
    int ret;
    int i;

    gw_assert(set != NULL);

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

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

        if (ret < 0) {
	    if (errno != EINTR) {
                error(0, "Poller: can't handle error; sleeping 1 second.");
                gwthread_sleep(1.0);
            }
            continue;
        }

	/* 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]);
        }
	set->scanning = 0;

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



FDSet *fdset_create(void)
{
    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->scanning = 0;
    new->deleted_entries = 0;

    new->actions = list_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;
}

void fdset_destroy(FDSet *set)
{
    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);
        if (list_len(set->actions) > 0) {
            error(0, "Destroying fdset with %ld pending actions.",
                  list_len(set->actions));
        }
        list_destroy(set->actions, action_destroy_item);
        gw_free(set);
    } else {
        long thread = set->poll_thread;
        submit_action(set, action_create(DESTROY));
	gwthread_join(thread);
    }
}

void fdset_register(FDSet *set, int fd, int events,
                    fdset_callback_t callback, void *data)
{
    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->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;
}

void fdset_listen(FDSet *set, int fd, int mask, int events)
{
    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);
    }
}

void fdset_unregister(FDSet *set, int fd)
{
    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);
    }
}