/* * event.c Non-thread-safe event handling, specific to a RADIUS * server. * * Version: $Id$ * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public * License as published by the Free Software Foundation; either * version 2.1 of the License, or (at your option) any later version. * * This library is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public * License along with this library; if not, write to the Free Software * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA * * Copyright 2007 The FreeRADIUS server project * Copyright 2007 Alan DeKok */ RCSID("$Id$") #include #include #include #include #ifdef HAVE_KQUEUE #ifndef HAVE_SYS_EVENT_H #error kqueue requires #else #include #endif #endif /* HAVE_KQUEUE */ typedef struct fr_event_fd_t { int fd; fr_event_fd_handler_t handler; fr_event_fd_handler_t write_handler; void *ctx; } fr_event_fd_t; #define FR_EV_MAX_FDS (512) #undef USEC #define USEC (1000000) struct fr_event_list_t { fr_heap_t *times; int exit; fr_event_status_t status; struct timeval now; bool dispatch; int num_readers; #ifndef HAVE_KQUEUE int max_readers; int max_fd; fd_set read_fds; fd_set write_fds; #else int kq; struct kevent events[FR_EV_MAX_FDS]; /* so it doesn't go on the stack every time */ #endif fr_event_fd_t readers[FR_EV_MAX_FDS]; }; /* * Internal structure for managing events. */ struct fr_event_t { fr_event_callback_t callback; void *ctx; struct timeval when; fr_event_t **parent; int heap; }; static int fr_event_list_time_cmp(void const *one, void const *two) { fr_event_t const *a = one; fr_event_t const *b = two; if (a->when.tv_sec < b->when.tv_sec) return -1; if (a->when.tv_sec > b->when.tv_sec) return +1; if (a->when.tv_usec < b->when.tv_usec) return -1; if (a->when.tv_usec > b->when.tv_usec) return +1; return 0; } static int _event_list_free(fr_event_list_t *list) { fr_event_list_t *el = list; fr_event_t *ev; while ((ev = fr_heap_peek(el->times)) != NULL) { fr_event_delete(el, &ev); } fr_heap_delete(el->times); #ifdef HAVE_KQUEUE close(el->kq); #endif return 0; } fr_event_list_t *fr_event_list_create(TALLOC_CTX *ctx, fr_event_status_t status) { int i; fr_event_list_t *el; el = talloc_zero(ctx, fr_event_list_t); if (!fr_assert(el)) { return NULL; } talloc_set_destructor(el, _event_list_free); el->times = fr_heap_create(fr_event_list_time_cmp, offsetof(fr_event_t, heap)); if (!el->times) { talloc_free(el); return NULL; } for (i = 0; i < FR_EV_MAX_FDS; i++) { el->readers[i].fd = -1; } #ifndef HAVE_KQUEUE el->max_fd = 0; FD_ZERO(&el->read_fds); FD_ZERO(&el->write_fds); #else el->kq = kqueue(); if (el->kq < 0) { talloc_free(el); return NULL; } #endif el->status = status; return el; } int fr_event_list_num_fds(fr_event_list_t *el) { if (!el) return 0; return el->num_readers; } int fr_event_list_num_elements(fr_event_list_t *el) { if (!el) return 0; return fr_heap_num_elements(el->times); } int fr_event_delete(fr_event_list_t *el, fr_event_t **parent) { int ret; fr_event_t *ev; if (!el || !parent || !*parent) return 0; #ifndef NDEBUG /* * Validate the event_t struct to detect memory issues early. */ ev = talloc_get_type_abort(*parent, fr_event_t); #else ev = *parent; #endif if (ev->parent) { fr_assert(*(ev->parent) == ev); *ev->parent = NULL; } *parent = NULL; ret = fr_heap_extract(el->times, ev); fr_assert(ret == 1); /* events MUST be in the heap */ talloc_free(ev); return ret; } int fr_event_insert(fr_event_list_t *el, fr_event_callback_t callback, void *ctx, struct timeval *when, fr_event_t **parent) { fr_event_t *ev; if (!el) { fr_strerror_printf("Invalid arguments (NULL event list)"); return 0; } if (!callback) { fr_strerror_printf("Invalid arguments (NULL callback)"); return 0; } if (!when || (when->tv_usec >= USEC)) { fr_strerror_printf("Invalid arguments (time)"); return 0; } if (!parent) { fr_strerror_printf("Invalid arguments (NULL parent)"); return 0; } /* * If there is an event, re-use it instead of freeing it * and allocating a new one. */ if (*parent) { int ret; #ifndef NDEBUG ev = talloc_get_type_abort(*parent, fr_event_t); #else ev = *parent; #endif ret = fr_heap_extract(el->times, ev); fr_assert(ret == 1); /* events MUST be in the heap */ memset(ev, 0, sizeof(*ev)); } else { ev = talloc_zero(el, fr_event_t); if (!ev) return 0; } ev->callback = callback; ev->ctx = ctx; ev->when = *when; ev->parent = parent; if (!fr_heap_insert(el->times, ev)) { talloc_free(ev); return 0; } *parent = ev; return 1; } int fr_event_run(fr_event_list_t *el, struct timeval *when) { fr_event_callback_t callback; void *ctx; fr_event_t *ev; if (!el) return 0; if (fr_heap_num_elements(el->times) == 0) { when->tv_sec = 0; when->tv_usec = 0; return 0; } ev = fr_heap_peek(el->times); if (!ev) { when->tv_sec = 0; when->tv_usec = 0; return 0; } #ifndef NDEBUG ev = talloc_get_type_abort(ev, fr_event_t); #endif /* * See if it's time to do this one. */ if ((ev->when.tv_sec > when->tv_sec) || ((ev->when.tv_sec == when->tv_sec) && (ev->when.tv_usec > when->tv_usec))) { *when = ev->when; return 0; } callback = ev->callback; ctx = ev->ctx; /* * Delete the event before calling it. */ fr_event_delete(el, ev->parent); callback(ctx); return 1; } int fr_event_now(fr_event_list_t *el, struct timeval *when) { if (!when) return 0; if (el && el->dispatch) { *when = el->now; } else { gettimeofday(when, NULL); } return 1; } int fr_event_fd_insert(fr_event_list_t *el, int type, int fd, fr_event_fd_handler_t handler, void *ctx) { int i; fr_event_fd_t *ef; if (!el) { fr_strerror_printf("Invalid arguments (NULL event list)"); return 0; } if (!handler) { fr_strerror_printf("Invalid arguments (NULL handler)"); return 0; } if (!ctx) { fr_strerror_printf("Invalid arguments (NULL ctx)"); return 0; } if (fd < 0) { fr_strerror_printf("Invalid arguments (bad FD %i)", fd); return 0; } if (type != 0) { fr_strerror_printf("Invalid type %i", type); return 0; } if (el->num_readers >= FR_EV_MAX_FDS) { fr_strerror_printf("Too many readers"); return 0; } ef = NULL; #ifdef HAVE_KQUEUE /* * We need to store TWO fields with the event. kqueue * only lets us store one. If we put the two fields into * a malloc'd structure, that would help. Except that * kqueue can silently delete the event when the socket * is closed, and not give us the opportunity to free it. * * * The solution is to put the fields into an array, and * do a linear search on addition/deletion of the FDs. * However, to avoid MOST linear issues, we start off the * search at "FD" offset. Since FDs are unique, AND * usually less than 256, we do "FD & 0xff", which is a * good guess, and makes the lookups mostly O(1). */ for (i = 0; i < FR_EV_MAX_FDS; i++) { int j; struct kevent evset; j = (i + fd) & (FR_EV_MAX_FDS - 1); if (el->readers[j].fd >= 0) continue; /* * We want to read from the FD. */ EV_SET(&evset, fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, &el->readers[j]); if (kevent(el->kq, &evset, 1, NULL, 0, NULL) < 0) { fr_strerror_printf("Failed inserting event for FD %i: %s", fd, fr_syserror(errno)); return 0; } ef = &el->readers[j]; el->num_readers++; break; } #else /* HAVE_KQUEUE */ /* * select() has limits. */ if (fd > FD_SETSIZE) { fprintf(stderr, "FD is larger than FD_SETSIZE"); return 0; } for (i = 0; i <= el->max_readers; i++) { /* * Be fail-safe on multiple inserts. */ if (el->readers[i].fd == fd) { if ((el->readers[i].handler != handler) || (el->readers[i].ctx != ctx)) { fr_strerror_printf("Multiple handlers for same FD"); return 0; } /* * No change. */ return 1; } if (el->readers[i].fd < 0) { ef = &el->readers[i]; el->num_readers++; if (i == el->max_readers) el->max_readers = i + 1; FD_SET(fd, &el->read_fds); if (el->max_fd <= fd) el->max_fd = fd; break; } } #endif if (!ef) { fr_strerror_printf("Failed assigning FD"); return 0; } ef->fd = fd; ef->handler = handler; ef->ctx = ctx; return 1; } int fr_event_fd_write_handler(fr_event_list_t *el, int type, int fd, fr_event_fd_handler_t write_handler, void *ctx) { int i; if (!el || (fd < 0)) return 0; if (type != 0) return 0; #ifdef HAVE_KQUEUE for (i = 0; i < FR_EV_MAX_FDS; i++) { int j; struct kevent evset; j = (i + fd) & (FR_EV_MAX_FDS - 1); if (el->readers[j].fd != fd) continue; fr_assert(ctx = el->readers[j].ctx); /* * Tell us when the socket is ready for writing */ if (write_handler) { fr_assert(!el->readers[j].write_handler); el->readers[j].write_handler = write_handler; EV_SET(&evset, fd, EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0, &el->readers[j]); } else { fr_assert(el->readers[j].write_handler); el->readers[j].write_handler = NULL; EV_SET(&evset, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL); } if (kevent(el->kq, &evset, 1, NULL, 0, NULL) < 0) { fr_strerror_printf("Failed inserting event for FD %i: %s", fd, fr_syserror(errno)); return 0; } return 1; } #else for (i = 0; i < el->max_readers; i++) { if (el->readers[i].fd != fd) continue; fr_assert(ctx = el->readers[i].ctx); el->readers[i].write_handler = write_handler; FD_SET(fd, &el->write_fds); /* fd MUST already be in the set of readers! */ return 1; } #endif /* HAVE_KQUEUE */ return 0; } int fr_event_fd_delete(fr_event_list_t *el, int type, int fd) { int i; if (!el || (fd < 0)) return 0; if (type != 0) return 0; #ifdef HAVE_KQUEUE for (i = 0; i < FR_EV_MAX_FDS; i++) { int j; struct kevent evset; j = (i + fd) & (FR_EV_MAX_FDS - 1); if (el->readers[j].fd != fd) continue; /* * Tell the kernel to delete it from the list. * * The caller MAY have closed it, in which case * the kernel has removed it from the list. So * we ignore the return code from kevent(). */ EV_SET(&evset, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL); (void) kevent(el->kq, &evset, 1, NULL, 0, NULL); /* * Delete the write handler if it exits. */ if (el->readers[j].write_handler) { EV_SET(&evset, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL); (void) kevent(el->kq, &evset, 1, NULL, 0, NULL); } el->readers[j].fd = -1; el->num_readers--; return 1; } #else for (i = 0; i < el->max_readers; i++) { if (el->readers[i].fd == fd) { el->readers[i].fd = -1; el->num_readers--; if ((i + 1) == el->max_readers) el->max_readers = i; FD_CLR(fd, &el->read_fds); FD_CLR(fd, &el->write_fds); /* * @todo - update el->max_fd, too. */ return 1; } } #endif /* HAVE_KQUEUE */ return 0; } void fr_event_loop_exit(fr_event_list_t *el, int code) { if (!el) return; el->exit = code; } bool fr_event_loop_exiting(fr_event_list_t *el) { return (el->exit != 0); } int fr_event_loop(fr_event_list_t *el) { int i, rcode; struct timeval when, *wake; #ifdef HAVE_KQUEUE struct timespec ts_when, *ts_wake; #else fd_set read_fds, write_fds; #endif el->exit = 0; el->dispatch = true; while (!el->exit) { /* * Find the first event. If there's none, we wait * on the socket forever. */ when.tv_sec = 0; when.tv_usec = 0; if (fr_heap_num_elements(el->times) > 0) { fr_event_t *ev; ev = fr_heap_peek(el->times); if (!ev) { fr_exit_now(42); } gettimeofday(&el->now, NULL); if (timercmp(&el->now, &ev->when, <)) { when = ev->when; when.tv_sec -= el->now.tv_sec; if (when.tv_sec > 0) { when.tv_sec--; when.tv_usec += USEC; } else { when.tv_sec = 0; } when.tv_usec -= el->now.tv_usec; if (when.tv_usec >= USEC) { when.tv_usec -= USEC; when.tv_sec++; } } else { /* we've passed the event time */ when.tv_sec = 0; when.tv_usec = 0; } wake = &when; } else { wake = NULL; } /* * Tell someone what the status is. */ if (el->status) el->status(wake); #ifndef HAVE_KQUEUE read_fds = el->read_fds; write_fds = el->write_fds; rcode = select(el->max_fd + 1, &read_fds, &write_fds, NULL, wake); if ((rcode < 0) && (errno != EINTR)) { fr_strerror_printf("Failed in select: %s", fr_syserror(errno)); el->dispatch = false; return -1; } #else /* HAVE_KQUEUE */ if (wake) { ts_wake = &ts_when; ts_when.tv_sec = when.tv_sec; ts_when.tv_nsec = when.tv_usec * 1000; } else { ts_wake = NULL; } rcode = kevent(el->kq, NULL, 0, el->events, FR_EV_MAX_FDS, ts_wake); #endif /* HAVE_KQUEUE */ if (fr_heap_num_elements(el->times) > 0) { do { gettimeofday(&el->now, NULL); when = el->now; } while (fr_event_run(el, &when) == 1); } if (rcode <= 0) continue; #ifndef HAVE_KQUEUE /* * Loop over all of the sockets to see if there's * an event for that socket. */ for (i = 0; i < el->max_readers; i++) { fr_event_fd_t *ef = &el->readers[i]; if (ef->fd < 0) continue; /* * Check if the socket is available for writing. */ if (ef->write_handler && FD_ISSET(ef->fd, &write_fds)) { ef->write_handler(el, ef->fd, ef->ctx); } if (!FD_ISSET(ef->fd, &read_fds)) continue; ef->handler(el, ef->fd, ef->ctx); } #else /* HAVE_KQUEUE */ /* * Loop over all of the events, servicing them. */ for (i = 0; i < rcode; i++) { fr_event_fd_t *ef = el->events[i].udata; if (el->events[i].flags & EV_EOF) { /* * FIXME: delete the handler * here, and fix process.c to not * call fr_event_fd_delete(). * It's cleaner. * * Call the handler, which SHOULD * delete the connection. */ ef->handler(el, ef->fd, ef->ctx); continue; } if (el->events[i].filter == EVFILT_WRITE) { ef->write_handler(el, ef->fd, ef->ctx); continue; } /* * Else it's our event. We only set * EVFILT_READ, so it must be a read * event. */ ef->handler(el, ef->fd, ef->ctx); } #endif /* HAVE_KQUEUE */ } el->dispatch = false; return el->exit; } #ifdef TESTING /* * cc -g -I .. -c rbtree.c -o rbtree.o && cc -g -I .. -c isaac.c -o isaac.o && cc -DTESTING -I .. -c event.c -o event_mine.o && cc event_mine.o rbtree.o isaac.o -o event * * ./event * * And hit CTRL-S to stop the output, CTRL-Q to continue. * It normally alternates printing the time and sleeping, * but when you hit CTRL-S/CTRL-Q, you should see a number * of events run right after each other. * * OR * * valgrind --tool=memcheck --leak-check=full --show-reachable=yes ./event */ static void print_time(void *ctx) { struct timeval *when = ctx; printf("%d.%06d\n", when->tv_sec, when->tv_usec); fflush(stdout); } static fr_randctx rand_pool; static uint32_t event_rand(void) { uint32_t num; num = rand_pool.randrsl[rand_pool.randcnt++ & 0xff]; if (rand_pool.randcnt == 256) { fr_isaac(&rand_pool); rand_pool.randcnt = 0; } return num; } #define MAX 100 int main(int argc, char **argv) { int i, rcode; struct timeval array[MAX]; struct timeval now, when; fr_event_list_t *el; el = fr_event_list_create(NULL, NULL); if (!el) exit(1); memset(&rand_pool, 0, sizeof(rand_pool)); rand_pool.randrsl[1] = time(NULL); fr_randinit(&rand_pool, 1); rand_pool.randcnt = 0; gettimeofday(&array[0], NULL); for (i = 1; i < MAX; i++) { array[i] = array[i - 1]; array[i].tv_usec += event_rand() & 0xffff; if (array[i].tv_usec > 1000000) { array[i].tv_usec -= 1000000; array[i].tv_sec++; } fr_event_insert(el, print_time, &array[i], &array[i]); } while (fr_event_list_num_elements(el)) { gettimeofday(&now, NULL); when = now; if (!fr_event_run(el, &when)) { int delay = (when.tv_sec - now.tv_sec) * 1000000; delay += when.tv_usec; delay -= now.tv_usec; printf("\tsleep %d\n", delay); fflush(stdout); usleep(delay); } } talloc_free(el); return 0; } #endif