path: root/lib/waiter/waiter.c

#include <poll.h>
#include <stdbool.h>
#include <string.h>
#include <assert.h>
#include <sys/time.h>

#include <talloc/talloc.h>

#include "waiter.h"

struct waiter {
	enum {
		WAITER_IO,
		WAITER_TIME,
	}		type;
	struct waitset	*set;
	int		fd;
	int		events;
	struct timeval	timeout;
	waiter_cb	callback;
	void		*arg;
};

struct waitset {
	struct waiter	**waiters;
	int		n_waiters;
	bool		waiters_changed;

	struct timeval	next_timeout;

	/* These are kept consistent over each call to waiter_poll, as
	 * set->waiters may be updated (by waiters' callbacks calling
	 * waiter_register or waiter_remove) during iteration. */
	struct pollfd	*pollfds;
	int		n_pollfds;
	struct waiter	**io_waiters;
	int		n_io_waiters;
	struct waiter	**time_waiters;
	int		n_time_waiters;
};

struct waitset *waitset_create(void *ctx)
{
	struct waitset *set = talloc_zero(ctx, struct waitset);
	return set;
}

void waitset_destroy(struct waitset *set)
{
	talloc_free(set);
}

static struct waiter *waiter_new(struct waitset *set)
{
	struct waiter **waiters, *waiter;

	waiter = talloc(set->waiters, struct waiter);
	if (!waiter)
		return NULL;

	waiters = talloc_realloc(set, set->waiters,
			struct waiter *, set->n_waiters + 1);

	if (!waiters) {
		talloc_free(waiter);
		return NULL;
	}

	set->waiters_changed = true;
	set->waiters = waiters;
	set->n_waiters++;

	set->waiters[set->n_waiters - 1] = waiter;
	return waiter;
}

struct waiter *waiter_register_io(struct waitset *set, int fd, int events,
		waiter_cb callback, void *arg)
{
	struct waiter *waiter = waiter_new(set);

	waiter->type = WAITER_IO;
	waiter->set = set;
	waiter->fd = fd;
	waiter->events = events;
	waiter->callback = callback;
	waiter->arg = arg;

	return waiter;
}

struct waiter *waiter_register_timeout(struct waitset *set, int delay_ms,
		waiter_cb callback, void *arg)
{
	struct waiter *waiter = waiter_new(set);
	struct timeval now, delay;

	delay.tv_sec = delay_ms / 1000;
	delay.tv_usec = 1000 * (delay_ms % 1000);

	gettimeofday(&now, NULL);

	timeradd(&now, &delay, &waiter->timeout);

	waiter->type = WAITER_TIME;
	waiter->set = set;
	waiter->callback = callback;
	waiter->arg = arg;

	return waiter;
}

void waiter_remove(struct waiter *waiter)
{
	struct waitset *set = waiter->set;
	int i;

	for (i = 0; i < set->n_waiters; i++)
		if (set->waiters[i] == waiter)
			break;

	assert(i < set->n_waiters);

	set->n_waiters--;
	memmove(&set->waiters[i], &set->waiters[i+1],
		(set->n_waiters - i) * sizeof(set->waiters[0]));

	set->waiters = talloc_realloc(set->waiters, set->waiters,
			struct waiter *, set->n_waiters);
	set->waiters_changed = true;

	talloc_free(waiter);
}

static void update_waiters(struct waitset *set)
{
	int n_io, n_time, i_io, i_time, i;

	if (!set->waiters_changed)
		return;

	n_io = n_time = 0;

	for (i = 0; i < set->n_waiters; i++) {
		if (set->waiters[i]->type == WAITER_IO)
			n_io++;
		else if (set->waiters[i]->type == WAITER_TIME)
			n_time++;
	}

	/* realloc if counts have changed */
	if (set->n_io_waiters != n_io) {
		set->io_waiters = talloc_realloc(set, set->io_waiters,
				struct waiter *, n_io);
		set->pollfds = talloc_realloc(set, set->pollfds,
				struct pollfd, n_io);
		set->n_io_waiters = n_io;
	}
	if (set->n_time_waiters != n_time) {
		set->time_waiters = talloc_realloc(set, set->time_waiters,
				struct waiter *, n_time);
		set->n_time_waiters = n_time;
	}

	/* IO waiters: copy to io_waiters, populate pollfds */
	for (i = i_io = 0; i < set->n_waiters; i++) {
		struct waiter *waiter = set->waiters[i];

		if (waiter->type != WAITER_IO)
			continue;

		set->pollfds[i_io].fd = waiter->fd;
		set->pollfds[i_io].events = waiter->events;
		set->io_waiters[i_io] = waiter;
		i_io++;
	}

	/* time waiters: copy to time_waiters, calculate next expiry */
	timerclear(&set->next_timeout);
	for (i = i_time = 0; i < set->n_waiters; i++) {
		struct waiter *waiter = set->waiters[i];

		if (waiter->type != WAITER_TIME)
			continue;

		if (!timerisset(&set->next_timeout) ||
				timercmp(&waiter->timeout,
					&set->next_timeout, <))
			set->next_timeout = waiter->timeout;

		set->time_waiters[i_time] = waiter;
		i_time++;
	}
}

int waiter_poll(struct waitset *set)
{
	struct timeval now, timeout;
	int timeout_ms;
	int i, rc;

	/* If the waiters have been updated, we need to update our
	 * consistent copy */
	update_waiters(set);

	if (timerisset(&set->next_timeout)) {
		gettimeofday(&now, NULL);
		timersub(&set->next_timeout, &now, &timeout);
		timeout_ms = timeout.tv_sec * 1000 +
				timeout.tv_usec / 1000;
		if (timeout_ms < 0)
			timeout_ms = 0;
	} else {
		timeout_ms = -1;
	}


	rc = poll(set->pollfds, set->n_io_waiters, timeout_ms);

	if (rc < 0)
		return rc;

	for (i = 0; i < set->n_io_waiters; i++) {
		struct waiter *waiter = set->io_waiters[i];

		if (!set->pollfds[i].revents)
			continue;
		rc = waiter->callback(waiter->arg);

		if (rc)
			waiter_remove(waiter);
	}

	if (set->n_time_waiters > 0)
		gettimeofday(&now, NULL);

	for (i = 0; i < set->n_time_waiters; i++) {
		struct waiter *waiter = set->time_waiters[i];

		if (timercmp(&waiter->timeout, &now, >))
			continue;

		waiter->callback(waiter->arg);

		waiter_remove(waiter);
	}

	return 0;
}