Add support for peridic timers and resetting timers

Refactor uev_timer_create() to support a periodic (msec) argument. For
such timers libuev automaticlly respawns them.  Effectively resetting
the timeout using the period value.

Also, refactor argument order to uev_timer_create() and uev_io_create().

Rename UEV_FHIN, UEV_FHOUT to UEV_DIR_INBOUND and UEV_DIR_OUTBOUND.

Remove unnecessary return value for watcher callbacks.  What would those
have been usefu for, ever?  Not even the original code checked the return
value after calling a handler.

Signed-off-by: Joachim Nilsson <troglobit@gmail.com>
This commit is contained in:
Joachim Nilsson committed 2013-07-21 11:39:26 +02:00
1 parent 97398e0a7d
commit ec9351bfdd
3 files changed
+140 -78

No files matched your search

+87 -34
View File
@@ -38,11 +38,19 @@
static clock_t clock_tick = 0;
uev_io_t *uev_io_create(uev_t *ctx, int fd, uev_dir_t dir, uev_io_cb_t handler, void *data)
static clock_t uev_time_now(uev_t *ctx)
{
if (!ctx->use_next)
return times(NULL);
return ctx->next_time;
}
uev_io_t *uev_io_create(uev_t *ctx, uev_io_cb_t handler, void *data, int fd, uev_dir_t dir)
{
uev_io_t *w;
w = (uev_io_t *)malloc(sizeof(*w));
w = (uev_io_t *)calloc(1, sizeof(*w));
if (!w)
return NULL;
@@ -80,50 +88,97 @@ int uev_io_delete(uev_t *ctx, uev_io_t *w)
return 0;
}
static int timer_active(uev_t *ctx, uev_timer_t *w)
{
uev_timer_t *tmp;
TAILQ_FOREACH(tmp, &ctx->timer_list, link) {
if (tmp == w)
return 1;
}
return 0;
}
/**
* Add a timer event
* @param ctx A valid libuev context
* @param handler Timer callback
* @param data Optional callback argument
* @param timeout Timeout in milliseconds before @param handler is called
* @param period For periodic timers this is the period time that @param timeout is reset to
*
* One-shot timers you likely set @param period to zero and only use
* @param timeout. For periodic timers you likely set @param timeout to
* either zero, to call it as soon as the event loop starts, or to the
* same value as @param period. When the timer expires, the @param
* handler is called, with the optional @param data argument. A
* non-periodic timer ends its life there, while a periodic task's
* @param timeout is reset to the @param period and restarted.
*
* @return The new timer, or %NULL if invalid pointers or out or memory.
*/
uev_timer_t *uev_timer_create(uev_t *ctx, int msec, uev_timer_cb_t handler, void *data)
uev_timer_t *uev_timer_create(uev_t *ctx, uev_timer_cb_t handler, void *data, int timeout, int period)
{
int diff;
clock_t now;
uev_timer_t *w, *ret, *tmp;
uev_timer_t *w;
if (!ctx) {
if (!ctx || !handler) {
errno = EINVAL;
return NULL;
}
w = (uev_timer_t *)malloc(sizeof(*w));
w = (uev_timer_t *)calloc(1, sizeof(*w));
if (!w)
return NULL;
w->due = msec;
w->handler = (void *)handler;
w->data = data;
/* Zero delay is a special case: A oneshot workproc: Must go first */
if (0 == msec) {
if (0 == timeout) {
TAILQ_INSERT_HEAD(&ctx->timer_list, w, link);
return w;
}
uev_timer_set(ctx, w, timeout, period);
return w;
}
/**
* Reset or reschedule a timer
*/
int uev_timer_set(uev_t *ctx, uev_timer_t *w, int timeout, int period)
{
int diff;
clock_t now;
uev_timer_t *tmp;
if (!ctx || !w) {
errno = EINVAL;
return -1;
}
w->timeout = timeout;
w->period = period;
/* Adjust timers.
* Jiffie wraparound is OK since we only care about diff time */
now = ctx->use_next ? ctx->next_time : times(NULL);
now = uev_time_now(ctx);
diff = TIME_DIFF_MSEC(now, ctx->base_time);
/* To be returned below */
ret = w;
/* When rescheduling we need to remove the entry first */
if (timer_active(ctx, w))
TAILQ_REMOVE(&ctx->timer_list, w, link);
/* Update all timers and add this entry (ordered) to the list */
TAILQ_FOREACH(tmp, &ctx->timer_list, link) {
if (tmp->due > diff)
tmp->due -= diff;
if (tmp->timeout > diff)
tmp->timeout -= diff;
else
tmp->due = 0;
tmp->timeout = 0;
if (w && tmp->due > msec) {
if (w && tmp->timeout > timeout) {
TAILQ_INSERT_BEFORE(tmp, w, link);
w = NULL;
}
@@ -135,7 +190,7 @@ uev_timer_t *uev_timer_create(uev_t *ctx, int msec, uev_timer_cb_t handler, void
/* Update time base */
ctx->base_time = now;
return ret;
return 0;
}
/**
@@ -177,15 +232,18 @@ static int do_run_pending(uev_t *ctx, int immediate)
/* Do due timers */
if (!TAILQ_EMPTY(&ctx->timer_list)) {
now = ctx->use_next ? ctx->next_time : times(NULL);
now = uev_time_now(ctx);
diff = TIME_DIFF_MSEC(now, ctx->base_time);
TAILQ_FOREACH(timer, &ctx->timer_list, link) {
if (timer->due > diff)
if (timer->timeout > diff)
continue;
timer->handler((struct uev *)ctx, timer, timer->data);
uev_timer_delete(ctx, timer);
if (!timer->period)
uev_timer_delete(ctx, timer);
else
uev_timer_set(ctx, timer, timer->period, timer->period);
/* base_time changes for every uev_timer_create(), which
* can be called in every timer->handler() above. */
@@ -202,8 +260,8 @@ static int do_run_pending(uev_t *ctx, int immediate)
/* Calculate waiting time */
if (ctx->exiting)
diff = 0;
else if ((timer = TAILQ_FIRST(&ctx->timer_list)) && timer->due > diff)
diff = timer->due - diff;
else if ((timer = TAILQ_FIRST(&ctx->timer_list)) && timer->timeout > diff)
diff = timer->timeout - diff;
else if (NULL != timer)
diff = 0;
else
@@ -221,7 +279,7 @@ static int do_run_pending(uev_t *ctx, int immediate)
TAILQ_FOREACH(w, &ctx->io_list, link) {
w->index = i;
polls[i].fd = w->fd;
polls[i].events = UEV_FHIN == w->dir ? POLLIN : POLLOUT;
polls[i].events = UEV_DIR_INBOUND == w->dir ? POLLIN : POLLOUT;
polls[i].revents = 0;
i++;
}
@@ -233,19 +291,19 @@ static int do_run_pending(uev_t *ctx, int immediate)
if (EBADF == errno) {
TAILQ_FOREACH(w, &ctx->io_list, link) {
if (fcntl(w->fd, F_GETFD) == -1 && EBADF == errno)
w->handler((struct uev *)ctx, w, w->fd, w->data);
w->handler((struct uev *)ctx, w, w->data);
}
} else {
/* Error in poll. Cannot continue */
assert(false);
return;
return -2;
}
}
if (result > 0) {
TAILQ_FOREACH(w, &ctx->io_list, link) {
if (w->handler && w->index >= 0 && polls[w->index].revents)
w->handler((struct uev *)ctx, w, w->fd, w->data);
w->handler((struct uev *)ctx, w, w->data);
}
}
@@ -333,7 +391,7 @@ uev_t *uev_ctx_create(void)
{
uev_t *ctx;
ctx = (uev_t *)malloc(sizeof(*ctx));
ctx = (uev_t *)calloc(1, sizeof(*ctx));
if (!ctx)
return NULL;
@@ -342,13 +400,8 @@ uev_t *uev_ctx_create(void)
TAILQ_INIT(&ctx->io_delist);
TAILQ_INIT(&ctx->timer_delist);
/* Get system time and clock resolution, in ticks/second */
ctx->base_time = times(NULL);
ctx->next_time = 0;
ctx->use_next = false;
ctx->exiting = false;
/* Get system clock resolution, in ticks/second */
if (0 == clock_tick)
clock_tick = sysconf(_SC_CLK_TCK);
+13 -11
View File
@@ -38,8 +38,8 @@ struct uev;
/* I/O direction */
typedef enum {
UEV_FHIN = 0,
UEV_FHOUT = 1
UEV_DIR_INBOUND = 0,
UEV_DIR_OUTBOUND = 1
} uev_dir_t;
/* I/O event watcher */
@@ -51,7 +51,7 @@ typedef struct uev_io {
uev_dir_t dir; ///< Direction: in or out
int index; ///< Index in poll array
int (*handler)(struct uev *, struct uev_io *, int, void *);
void (*handler)(struct uev *, struct uev_io *, void *);
void *data;
} uev_io_t;
@@ -59,9 +59,10 @@ typedef struct uev_io {
typedef struct uev_timer {
TAILQ_ENTRY(uev_timer) link;
int due;
int timeout; ///< Timeout in milliseconds
int period; ///< Period time, in milliseconds
int (*handler)(struct uev *, struct uev_timer *, void *);
void (*handler)(struct uev *, struct uev_timer *, void *);
void *data;
} uev_timer_t;
@@ -81,15 +82,16 @@ typedef struct {
} uev_t;
/* Callbacks for different watchers */
typedef int (*uev_io_cb_t) (uev_t *ctx, uev_io_t *w, int fd, void *data);
typedef int (*uev_timer_cb_t)(uev_t *ctx, uev_timer_t *w, void *data);
typedef void (*uev_io_cb_t) (uev_t *ctx, uev_io_t *w, int fd, void *data);
typedef void (*uev_timer_cb_t)(uev_t *ctx, uev_timer_t *w, void *data);
/* Public interface */
uev_io_t *uev_io_create (uev_t *ctx, int fd, uev_dir_t dir, uev_io_cb_t handler, void *data);
int uev_io_delete (uev_t *ctx, uev_io_t *io);
uev_io_t *uev_io_create (uev_t *ctx, uev_io_cb_t cb, void *data, int fd, uev_dir_t dir);
int uev_io_delete (uev_t *ctx, uev_io_t *w);
uev_timer_t *uev_timer_create (uev_t *ctx, int msec, uev_timer_cb_t handler, void *data);
int uev_timer_delete (uev_t *ctx, uev_timer_t *hdl);
int uev_timer_set (uev_t *ctx, uev_timer_t *w, int timeout, int period);
uev_timer_t *uev_timer_create (uev_t *ctx, uev_timer_cb_t cb, void *data, int timeout, int period);
int uev_timer_delete (uev_t *ctx, uev_timer_t *w);
void uev_run (uev_t *ctx);
void uev_run_tick (uev_t *ctx, int msec);
+40 -33
View File
@@ -33,57 +33,56 @@ static int in, out;
static int period = 0;
static uev_timer_t *timer = NULL;
static int lifetime_cb(uev_t *ctx, uev_timer_t *w, void *data)
static void lifetime_cb(uev_t *ctx, uev_timer_t *w __attribute__ ((unused)), void *data)
{
fprintf(stderr, "\nLifetime exceeded %p\n", data);
uev_exit(ctx);
return 0;
}
/* The pipe watchdog, if it triggers we haven't received data in time. */
static int timeout_cb(uev_t *ctx, uev_timer_t *w, void *data)
static void timeout_cb(uev_t *ctx, uev_timer_t *w, void *data)
{
timer = NULL;
fprintf(stderr, "\nTimeout exceeded %p\n", data);
uev_timer_delete(ctx, w);
uev_exit(ctx);
return 0;
}
static int pipe_read_cb(uev_t *ctx, uev_io_t *w, int fd, void *data)
static void periodic_task(uev_t *ctx __attribute__ ((unused)), uev_timer_t *w __attribute__ ((unused)), void *data __attribute__ ((unused)))
{
fprintf(stderr, "|");
}
static void pipe_read_cb(uev_t *ctx, uev_io_t *w __attribute__ ((unused)), void *data __attribute__ ((unused)))
{
int cnt;
char msg[50];
/* Kick watchdog */
if (timer) {
uev_timer_delete(ctx, timer);
timer = uev_timer_create(ctx, 1000, timeout_cb, NULL);
}
if (timer)
uev_timer_set(ctx, timer, 1000, 0);
cnt = read(in, msg, sizeof(msg));
// fprintf(stderr, "READ %.*s %d\n", cnt, msg, cnt);
fprintf(stderr, "%.*s.%d ", cnt, msg, cnt);
return 0;
}
static int pipe_write_cb(uev_t *ctx, uev_timer_t *handle, void *data)
static void pipe_write_cb(uev_t *ctx, uev_timer_t *w __attribute__ ((unused)), void *data)
{
int cnt = (int)(intptr_t)data;
char *msg = "TESTING";
write(out, msg, cnt);
if (write(out, msg, cnt) < 0) {
perror("\nFailed writing to pipe");
return;
}
// fprintf(stderr, "WRITE %.*s %d\n", cnt, msg, cnt);
fprintf(stderr, "%d ", cnt);
period = cnt + 5;
uev_timer_create(ctx, period * 100, pipe_write_cb, (void *)(intptr_t)(cnt + 1));
return 0;
uev_timer_create(ctx, pipe_write_cb, (void *)(intptr_t)(cnt + 1), period * 100, 0);
}
int main(void)
@@ -91,31 +90,39 @@ int main(void)
int fd[2];
uev_t *ctx = uev_ctx_create();
/* Total program execution time */
uev_timer_create(ctx, 4000, lifetime_cb, (void *)(intptr_t)2);
/* Total program execution time */
uev_timer_create(ctx, (uev_timer_cb_t)lifetime_cb, (void *)(intptr_t)2, 4000, 0);
/* Work load, one timer callback writes to a pipe periodically,
* and one I/O watcher that is called every time the pipe has
* some data. */
if (pipe(fd) < 0) {
/* Cleanup */
uev_ctx_delete(ctx);
return 1;
}
/* Work load, one timer callback writes to a pipe periodically,
* and one I/O watcher that is called every time the pipe has
* some data. */
pipe(fd);
in = fd[0];
out = fd[1];
uev_timer_create(ctx, 400, pipe_write_cb, (void *)(intptr_t)1);
uev_io_create(ctx, in, UEV_FHIN, pipe_read_cb, NULL);
uev_timer_create(ctx, (uev_timer_cb_t)pipe_write_cb, (void *)(intptr_t)1, 400, 0);
uev_io_create(ctx, (uev_io_cb_t)pipe_read_cb, NULL, in, UEV_DIR_INBOUND);
/* Watchdog for the above timer callback, if it doesn't wake up
* and write to the pipe within a given deadline it will bark. */
timer = uev_timer_create(ctx, 950, timeout_cb, (void *)(intptr_t)1);
/* Watchdog for the above timer callback, if it doesn't wake up
* and write to the pipe within a given deadline it will bark. */
timer = uev_timer_create(ctx, (uev_timer_cb_t)timeout_cb, (void *)(intptr_t)1, 950, 0);
/* Start event loop */
/* Periodic background task */
uev_timer_create(ctx, (uev_timer_cb_t)periodic_task, NULL, 200, 200);
/* Start event loop */
uev_run(ctx);
/* Tear down event context */
/* Tear down event context */
uev_ctx_delete(ctx);
fprintf(stderr, "Period is %d must be 10: %s\n", period, 10 == period ? "OK" : "ERROR!");
if (10 != period)
return 1;
if (10 != period)
return 1;
return 0;
}