commit: r412 - in branches/events_testing: daemon include
[email protected] Fri, 15 May 2009 17:54:18 -0400
| Newsgroups | gmane.network.spread.cvs |
|---|---|
| Message-ID | <[email protected]> |
Author: jschultz Date: 2009-05-15 17:54:18 -0400 (Fri, 15 May 2009) New Revision: 412 Modified: branches/events_testing/daemon/events.c branches/events_testing/include/sp_events.h Log: New implementation of events with a slightly cleaned up and expanded interface. Modified: branches/events_testing/daemon/events.c =================================================================== --- branches/events_testing/daemon/events.c 2009-05-04 15:21:16 UTC (rev 411) +++ branches/events_testing/daemon/events.c 2009-05-15 21:54:18 UTC (rev 412) @@ -1,764 +1,2508 @@ -/* - * The Spread Toolkit. - * - * The contents of this file are subject to the Spread Open-Source - * License, Version 1.0 (the ``License''); you may not use - * this file except in compliance with the License. You may obtain a - * copy of the License at: - * - * http://www.spread.org/license/ - * - * or in the file ``license.txt'' found in this distribution. - * - * Software distributed under the License is distributed on an AS IS basis, - * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License - * for the specific language governing rights and limitations under the - * License. - * - * The Creators of Spread are: - * Yair Amir, Michal Miskin-Amir, Jonathan Stanton, John Schultz. - * - * Copyright (C) 1993-2006 Spread Concepts LLC <[email protected]> - * - * All Rights Reserved. - * - * Major Contributor(s): - * --------------- - * Ryan Caudy [email protected] - contributions to process groups. - * Claudiu Danilov [email protected] - scalable wide area support. - * Cristina Nita-Rotaru [email protected] - group communication security. - * Theo Schlossnagle [email protected] - Perl, autoconf, old skiplist. - * Dan Schoenblum [email protected] - Java interface. - * - */ +#include "arch.h" +/*#define HAVE_LINUX*/ -#include "arch.h" +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <errno.h> +#include <assert.h> +#include <time.h> -/* undef redefined variables under windows */ #ifdef ARCH_PC_WIN95 -#undef EINTR -#undef EAGAIN -#undef EWOULDBLOCK -#undef EINPROGRESS + +# include <winsock.h> +# include <sys/timeb.h> + +#else + +# include <sys/time.h> +# include <sys/types.h> +# include <unistd.h> + +# ifdef HAVE_LINUX +# include <sys/epoll.h> +# endif + #endif -#include <errno.h> -#ifndef ARCH_PC_WIN95 +#include <stdutil/stddefines.h> +#include <stdutil/stddll.h> +#include <stdutil/stdskl.h> -#include <time.h> -#include <sys/time.h> -#include <sys/types.h> -#include <unistd.h> -#else /* ARCH_PC_WIN95 */ +#include <sp_events.h> -#include <winsock.h> -#include <sys/timeb.h> +/*************************************************************************************************************************** + ***************************************************************************************************************************/ -#endif /* ARCH_PC_WIN95 */ +#define E_MIN_PRIORITY 0 -#include <string.h> -#include "sp_events.h" -#include "objects.h" /* For memory */ -#include "memory.h" /* for memory */ -#include "alarm.h" +#define E_DELTA_TIMER 3 +#define E_ABSOLUTE_TIMER 4 +#define E_PERIODIC_TIMER 5 -typedef struct dummy_t_event { - sp_time t; - void (* func)( int code, void *data ); - int code; - void *data; - struct dummy_t_event *next; -} time_event; +#define E_EPOLL_INITIAL_SIZE 32 -typedef struct dummy_fd_event { - int fd; - int fd_type; - void (* func)( mailbox mbox, int code, void *data ); - int code; - void *data; - int active; /* true if active, false if inactive */ -} fd_event; +enum +{ + E_BUG = INT_MIN, + E_NOT_IMPLEMENTED, + E_NO_SYS, + E_SYS_FAILURE, + E_ALLOC_FAILURE, + E_INTERRUPTED, + E_TOO_MANY_FILES, + E_INVALID_PARAM, + E_DEADEND, -typedef struct dummy_fd_queue { - int num_fds; - int num_active_fds; - fd_event events[MAX_FD_EVENTS]; -} fd_queue; + E_UNKNOWN = -1, + E_REPOPULATE = 1, -static time_event *Time_queue; -static sp_time Now; +}; -static fd_queue Fd_queue[NUM_PRIORITY]; -static fd_set Fd_mask[NUM_FDTYPES]; -static int Active_priority; -static int Exit_events; +typedef enum +{ + E_BACKEND_UNSPECIFIED, + E_BACKEND_SELECT, + E_BACKEND_EPOLL, -int E_init(void) +} E_backend_type; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef stdint32 E_watch_id; +typedef int E_watch_type; +typedef int E_priority; +typedef void (*E_callback_fcn)(); +typedef void (*E_fd_callback_fcn)(int fd, int code, void *data); +typedef void (*E_time_callback_fcn)(int code, void *data); + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef struct { - int i,ret; - - Time_queue = NULL; + E_watch_type watch_type; + E_priority priority; - ret = Mem_init_object(TIME_EVENT, sizeof(time_event), 100,0); - if (ret < 0) - { - Alarm(EXIT, "E_Init: Failure to Initialize TIME_EVENT memory objects\n"); - } + E_callback_fcn callback_fcn; + int callback_code; + void * callback_data; - for ( i=0; i < NUM_PRIORITY; i++ ) - { - Fd_queue[i].num_fds = 0; - Fd_queue[i].num_active_fds = 0; - } - for ( i=0; i < NUM_FDTYPES; i++ ) - { - FD_ZERO( &Fd_mask[i] ); - } - Active_priority = LOW_PRIORITY; + stdbool noticed; /* if this watch has triggered and generated a notice, then notice_it is the */ + stdit notice_it; /* position in events->priorities[priority].notices */ - E_get_time(); +} E_watch; - Alarm( EVENTS, "E_init: went ok\n"); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - return( 0 ); +typedef struct +{ + E_watch w; /* NOTE: must remain first member, so we can freely cast between E_watch* and E_time_watch* */ + + sp_time timeout; + + stdit time_watches_it; /* position in events->time_watches */ + + stdbool in_timeouts; /* whether or not in events->timeouts; if so, then timeouts_it is the */ + stdit timeouts_it; /* position in events->timeouts */ + +} E_time_watch; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef struct +{ + E_watch w; /* NOTE: must remain first member, so we can freely cast between E_watch* and E_fd_watch* */ + + int fd; + stdbool active; /* whether or not this fd watch is currently active (i.e. - watched/queried) or not */ + stdit fd_watches_it; /* position in events->fd_watches */ + +} E_fd_watch; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef struct +{ + E_priority priority; + stddll notices; /* <E_watch*>: watches at this priority that have triggered / been noticed */ + + unsigned starve_thresh; /* threshold that determines if/when this priority elevates to "preempt" execution */ + unsigned starve_cnter; /* increments once for every non-preempt notice dispatch from a different priority */ + + int num_active_fd_watches; + + void * backend_lvl_info; + +} E_priority_info; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +#ifdef HAVE_LINUX + +typedef struct +{ + struct epoll_event * events; + int num_events; + +} E_epoll_info; + +typedef struct +{ + int epoll_fd; + int num_poll_fds; + +} E_epoll_lvl_info; + +#endif + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef struct +{ + fd_set tmp_fd_sets[NUM_FDTYPES]; + +} E_select_info; + +typedef struct +{ + fd_set fd_sets[NUM_FDTYPES]; + stdskl fd_skls[NUM_FDTYPES]; /* <int -> nil>: ordered by E_int_cmp */ + +} E_select_lvl_info; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +typedef struct +{ + stdbool keep_running; /* whether or not the current call to E_events_handle_events() should keep looping */ + E_priority max_priority; /* highest priority lvl in the system */ + E_priority active_threshold; /* minimum priority lvl that is currently being monitored */ + + sp_time last_clock_time; /* most recent time returned from the clock for this system */ + sp_time last_loop_time; /* most recent time when we returned from an OS query */ + + stdskl time_watches; /* <E_time_watch* -> nil>: ordered by E_time_watch_ptr_cmp: [fcn, code, data] */ + stdskl fd_watches; /* <int -> E_fd_watch*>: clustered, not ordered, by [fd] */ + + E_priority_info * priorities; /* array of size max_priority: notification Qs and other per priority lvl info */ + + /* notification generation mechanisms */ + + stdskl timeouts; /* <sp_time -> E_time_watch*>: ordered by absolute time out */ + + E_backend_type backend_type; /* method used by this sytem to query fds */ + void * backend_info; /* backend specific info for this system */ + +} E_events; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_int_cmp(const void * left, const void * right); +static int E_time_watch_ptr_cmp(const void * left, const void * right); +static int E_sp_time_cmp(const void * left, const void * right); + +static E_fd_watch * E_events_get_fd_watch(E_events * events, int fd, E_watch_type fd_type, stdit * sit); + +static sp_time E_events_get_time(E_events * events); +static sp_time E_events_get_loop_time(E_events * events); +static sp_time E_events_update_loop_time(E_events * events); + +static int E_events_init(E_events * events); +static void E_events_fini(E_events * events); + +static void E_events_exit_events(E_events * events); +static int E_events_handle_events(E_events * events); +static int E_events_populate_priority_queues(E_events * events, E_priority dispatch_lvl); +static int E_events_handle_starvation(E_events * events); +static int E_events_dispatch_notice(E_events * events, E_priority dispatch_lvl); + +static int E_events_create_timer(E_events * e, sp_time TO, E_watch_type t, E_time_callback_fcn f, int code, void * d, E_priority p); +static int E_events_attach_fd(E_events * e, int fd, E_watch_type t, E_fd_callback_fcn f, int code, void * d, E_priority p); +static void E_events_cancel_watch(E_events * events, E_watch * watch); + +static int E_events_queue(E_events * events, E_time_callback_fcn fcn, int code, void *data, sp_time delta_time); +static int E_events_queue_absolute(E_events * events, E_time_callback_fcn fcn, int code, void *data, sp_time absolute_time); +static int E_events_queue_periodic(E_events * events, E_time_callback_fcn fcn, int code, void *data, sp_time periodic_time); +static int E_events_dequeue(E_events * events, E_time_callback_fcn fcn, int code, void *data); +static int E_events_detach_fd(E_events * events, int fd, E_watch_type fd_type); + +static int E_events_set_active_threshold(E_events * events, E_priority priority); +static int E_events_num_active(E_events * events, E_priority priority); +static int E_events_activate_fd(E_events * events, int fd, E_watch_type fd_type); +static int E_events_deactivate_fd(E_events * events, int fd, E_watch_type fd_type); + +static int E_events_set_elevate_count(E_events * events, E_priority priority, unsigned count); +static int E_events_get_elevate_count(E_events * events, E_priority priority, unsigned * count); + +static int E_events_select_init(E_events * events); +static void E_events_select_fini(E_events * events); +static int E_events_select_activate_watch(E_events * events, E_fd_watch * fd_watch); +static int E_events_select_deactivate_watch(E_events * events, E_fd_watch * fd_watch); +static int E_events_select_query(E_events * events, E_priority query_lvl, const sp_time * delta_timeout, E_priority * queried_lvl); + +static int E_events_epoll_init(E_events * events); +static void E_events_epoll_fini(E_events * events); +static int E_events_epoll_activate_watch(E_events * events, E_fd_watch * fd_watch); +static int E_events_epoll_deactivate_watch(E_events * events, E_fd_watch * fd_watch); +static int E_events_epoll_query(E_events * events, E_priority query_lvl, const sp_time * delta_timeout, E_priority * queried_lvl); + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static E_events Events; /* implicit event system for public E_* calls */ + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_int_cmp(const void * left, + const void * right) +{ + return *(int*) left - *(int*) right; } -sp_time E_get_time(void) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_time_watch_ptr_cmp(const void * left, + const void * right) { -#ifndef ARCH_PC_WIN95 - struct timeval read_time; + E_time_watch * left_watch = *(E_time_watch**) left; + E_time_watch * right_watch = *(E_time_watch**) right; + int ret = 0; -#if HAVE_STRUCT_TIMEZONE - struct timezone dummy_tz; + if (left_watch->w.callback_fcn < right_watch->w.callback_fcn) { ret = -1; + } else if (left_watch->w.callback_fcn > right_watch->w.callback_fcn) { ret = 1; + } else if (left_watch->w.callback_code < right_watch->w.callback_code) { ret = -1; + } else if (left_watch->w.callback_code > right_watch->w.callback_code) { ret = 1; + } else if (left_watch->w.callback_data < right_watch->w.callback_data) { ret = -1; + } else if (left_watch->w.callback_data > right_watch->w.callback_data) { ret = 1; + } + + return ret; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_sp_time_cmp(const void * left, + const void * right) +{ + return E_compare_time(*(const sp_time*) left, *(const sp_time*) right); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static E_fd_watch * E_events_get_fd_watch(E_events * events, + int fd, + E_watch_type fd_type, + stdit * sit) +{ + E_fd_watch * fd_watch; + + for (stdskl_lowerb(&events->fd_watches, sit, &fd); !stdskl_is_end(&events->fd_watches, sit); stdskl_it_next(sit)) { + + fd_watch = *(E_fd_watch**) stdskl_it_val(sit); + + if (fd_watch->fd != fd) { /* no such watch */ + break; + } + + if (fd_watch->w.watch_type == fd_type) { /* exact match */ + return fd_watch; + } + } + + return NULL; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static sp_time E_events_get_time(E_events * events) +#ifdef ARCH_PC_WIN95 +{ + struct _timeb t; + + _ftime(&t); + + events->last_clock_time.sec = t.time; + events->last_clock_time.usec = (long) t.millitm * 1000; + + return events->last_clock_time; +} #else - sp_time dummy_tz; +{ + struct timeval t; + + gettimeofday(&t, NULL); + + events->last_clock_time.sec = t.tv_sec; + events->last_clock_time.usec = t.tv_usec; + + return events->last_clock_time; +} #endif - int ret; - ret = gettimeofday( &read_time, &dummy_tz ); - if ( ret < 0 ) Alarm( EXIT, "E_get_time: gettimeofday problems.\n" ); - Now.sec = read_time.tv_sec; - Now.usec = read_time.tv_usec; +/*************************************************************************************************************************** + ***************************************************************************************************************************/ -#else /* ARCH_PC_WIN95 */ +static sp_time E_events_get_loop_time(E_events * events) +{ + return events->last_loop_time; +} - struct _timeb timebuffer; +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - _ftime( &timebuffer ); +static sp_time E_events_update_loop_time(E_events * events) +{ + events->last_loop_time = E_events_get_time(events); - Now.sec = timebuffer.time; - Now.usec= timebuffer.millitm; - Now.usec= Now.usec * 1000; + return events->last_loop_time; +} -#endif /* ARCH_PC_WIN95 */ -#if 0 - Alarm( EVENTS, "E_get_time: time is (%d, %d)\n", Now.sec, Now.usec); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_init(E_events * events) +{ + E_priority_info * info; + E_priority priority; + int ret = 0; + + memset(events, 0, sizeof(*events)); + + events->keep_running = STDFALSE; /* NOTE: only set in E_events_handle_events; indicates loop is running */ + events->max_priority = HIGH_PRIORITY; + events->active_threshold = LOW_PRIORITY; + + E_events_update_loop_time(events); /* sets events->last_clock_time and events->last_loop_time */ + + if (stdskl_construct(&events->time_watches, sizeof(E_time_watch*), 0, E_time_watch_ptr_cmp) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + if (stdskl_construct(&events->fd_watches, sizeof(int), sizeof(E_fd_watch*), NULL) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL_TIME_WATCHES; + } + + if ((events->priorities = (E_priority_info*) calloc(events->max_priority + 1, sizeof(E_priority_info))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL_FD_WATCHES; + } + + for (priority = E_MIN_PRIORITY; priority <= events->max_priority; ++priority) { + + info = &events->priorities[priority]; + info->priority = priority; + + if (stddll_construct(&info->notices, sizeof(E_watch*)) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL_PRIORITIES; + } + + info->starve_thresh = 0; + info->starve_cnter = 0; + info->num_active_fd_watches = 0; + info->backend_lvl_info = NULL; + } + + if (stdskl_construct(&events->timeouts, sizeof(sp_time), sizeof(E_time_watch*), E_sp_time_cmp) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL_PRIORITIES; + } + + events->backend_type = E_BACKEND_UNSPECIFIED; + +#ifdef HAVE_LINUX + ret = E_events_epoll_init(events); +#else + ret = E_events_select_init(events); #endif - return ( Now ); + + if (ret != 0) { + goto FAIL_TIMEOUTS; + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL_TIMEOUTS: + stdskl_destruct(&events->timeouts); + + FAIL_PRIORITIES: + while (priority-- != E_MIN_PRIORITY) { + stddll_destruct(&events->priorities[priority].notices); + } + + free(events->priorities); + + FAIL_FD_WATCHES: + stdskl_destruct(&events->fd_watches); + + FAIL_TIME_WATCHES: + stdskl_destruct(&events->time_watches); + + FAIL: + assert(ret != 0); + + END: + return ret; } -sp_time E_sub_time( sp_time t, sp_time delta_t ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static void E_events_fini(E_events * events) { - sp_time res; + E_priority priority; + stdit sit; + +#ifdef HAVE_LINUX + E_events_epoll_fini(events); +#else + E_events_select_fini(events); +#endif - res.sec = t.sec - delta_t.sec; - res.usec = t.usec - delta_t.usec; - if ( res.usec < 0 ) - { - res.usec = res.usec + 1000000; - res.sec--; - } - if ( res.sec < 0 ) Alarm( EVENTS, "E_sub_time: negative time result.\n"); - return ( res ); + stdskl_destruct(&events->timeouts); + + for (priority = E_MIN_PRIORITY; priority <= events->max_priority; ++priority) { + stddll_destruct(&events->priorities[priority].notices); + } + + free(events->priorities); + + for (stdskl_begin(&events->fd_watches, &sit); !stdskl_is_end(&events->fd_watches, &sit); stdskl_it_next(&sit)) { + free(*(E_fd_watch**) stdskl_it_val(&sit)); + } + + stdskl_destruct(&events->fd_watches); + + for (stdskl_begin(&events->time_watches, &sit); !stdskl_is_end(&events->time_watches, &sit); stdskl_it_next(&sit)) { + free(*(E_time_watch**) stdskl_it_key(&sit)); + } + + stdskl_destruct(&events->time_watches); + memset(&events, 0, sizeof(*events)); } -sp_time E_add_time( sp_time t, sp_time delta_t ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static void E_events_exit_events(E_events * events) { - sp_time res; + events->keep_running = STDFALSE; +} - res.sec = t.sec + delta_t.sec; - res.usec = t.usec + delta_t.usec; - if ( res.usec > 1000000 ) - { - res.usec = res.usec - 1000000; - res.sec++; - } - return ( res ); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_handle_events(E_events * events) +{ + int ret = 0; + E_priority dispatch_lvl = events->max_priority; + E_priority starve_lvl; + + events->keep_running = STDTRUE; + + do { + + /* efficiently query watches for new notices */ + + if ((ret = E_events_populate_priority_queues(events, dispatch_lvl)) != 0) { + goto FAIL; + } + + /* handle any "left over" starvation from previous iterations */ + + if ((ret = E_events_handle_starvation(events)) != 0) { + + if (ret == E_REPOPULATE) { /* need to repopulate priority queues to ensure we aren't starving empty levels */ + ret = 0; + goto CONTINUE_LOOP; /* NOTE: should only happen here if in a starved callback the user lowered active_threshold */ + } + + goto FAIL; + } + + /* find the highest non-empty priority */ + + for (dispatch_lvl = events->max_priority; + dispatch_lvl >= events->active_threshold && stddll_empty(&events->priorities[dispatch_lvl].notices); + --dispatch_lvl); + + /* dispatch some notices from dispatch_lvl */ + + while (events->keep_running && dispatch_lvl >= events->active_threshold && !stddll_empty(&events->priorities[dispatch_lvl].notices)) { + + if ((ret = E_events_dispatch_notice(events, dispatch_lvl)) != 0) { + goto FAIL; + } + + /* increment starvation counts for other priorities */ + + for (starve_lvl = events->max_priority; starve_lvl >= events->active_threshold; --starve_lvl) { + + if (events->priorities[starve_lvl].starve_thresh != 0 && starve_lvl != dispatch_lvl) { + ++events->priorities[starve_lvl].starve_cnter; + } + } + + /* handle any starvation */ + + if ((ret = E_events_handle_starvation(events)) != 0) { + + if (ret == E_REPOPULATE) { /* need to repopulate priority queues to ensure we aren't starving empty levels */ + ret = 0; + goto CONTINUE_LOOP; + } + + goto FAIL; + } + } + + CONTINUE_LOOP:; + + } while (events->keep_running); + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0 && ret != E_REPOPULATE); + + END: + return ret; } -int E_compare_time( sp_time t1, sp_time t2 ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_populate_priority_queues(E_events * events, + E_priority dispatch_lvl) { - if ( t1.sec > t2.sec ) return ( 1 ); - else if ( t1.sec < t2.sec ) return ( -1 ); - else if ( t1.usec > t2.usec ) return ( 1 ); - else if ( t1.usec < t2.usec ) return ( -1 ); - else return ( 0 ); + E_priority starve_lvl; + E_priority query_lvl; + E_priority queried_lvl = events->max_priority; + E_watch * watch; + sp_time timeout; + sp_time * timeout_ptr = &timeout; + sp_time * abs_timeout; + E_priority tmp_lvl; + stdit sit; + stdit qit; + int ret; + + /* We postpone querying a non-empty priority level's (and all lower + levels too) fds until its dispatch queue is completey drained. + This ensures we do not query fds (and process any redundant + results) for which notifications are already waiting in the + dispatch queues. This approach also allows us to use the + smallest query sets possible, which often speeds up queries. + + However, we may lower the level to which we query in order to + check for potentially starving notices at those lower levels. We + may also lower the level to which we query for performance + reasons. For example, select()ing on 10 fds is only 30% more + expensive than selecting on 1 fd. So, it may make sense for + certain backend implementations to further lower the query level + in certain cases. + */ + + /* find the lowest priority lvl that we must query due to potential starvation */ + + for (starve_lvl = tmp_lvl = events->max_priority; tmp_lvl >= events->active_threshold; --tmp_lvl) { + + if (events->priorities[tmp_lvl].starve_thresh != 0 && /* starvation protection enabled */ + events->priorities[tmp_lvl].starve_cnter >= events->priorities[tmp_lvl].starve_thresh && /* starvation threshold reached */ + stddll_empty(&events->priorities[tmp_lvl].notices)) { /* empty -> potential starvation */ + + starve_lvl = tmp_lvl; + } + } + + /* dispatch_lvl was the highest non-empty priority the last time we dispatched notices */ + + /* NOTE: dispatch_lvl can go below active_threshold if user changes + the threshold in a callback; can also go to E_MIN_PRIORITY - 1 + when the active_threshold is E_MIN_PRIORITY and there are no + notices to dispatch + */ + + /* if we didn't drain dispatch_lvl then just query higher events->priorities */ + /* query the minimum between that and what starvation requires */ + + assert(dispatch_lvl <= events->max_priority && dispatch_lvl >= E_MIN_PRIORITY - 1); + + if (dispatch_lvl < events->active_threshold) { + query_lvl = events->active_threshold; + + } else { + query_lvl = (!stddll_empty(&events->priorities[dispatch_lvl].notices) ? dispatch_lvl + 1 : dispatch_lvl); + } + + query_lvl = STDMIN(query_lvl, starve_lvl); + + assert(query_lvl <= events->max_priority && query_lvl >= events->active_threshold && query_lvl >= E_MIN_PRIORITY); + + /* figure out an appropriate (delta) timeout and query backend implementation */ + + do { + + /* if we are querying a priority lvl higher than the lowest possible -> use a zero timeout */ + + if (query_lvl != events->active_threshold) { + + timeout_ptr->sec = 0; + timeout_ptr->usec = 0; + + } else if (!stdskl_empty(&events->timeouts)) { /* else calcualte an appropriate timeout */ + + abs_timeout = (sp_time*) stdskl_it_key(stdskl_begin(&events->timeouts, &sit)); + + E_events_update_loop_time(events); + + timeout_ptr->sec = abs_timeout->sec - events->last_loop_time.sec; + timeout_ptr->usec = abs_timeout->usec - events->last_loop_time.usec; + + if (timeout_ptr->usec < 0) { + timeout_ptr->usec += 1000000; + --timeout_ptr->sec; + } + + assert(timeout_ptr->usec >= 0 && timeout_ptr->usec < 1000000); + + if (timeout_ptr->sec < 0) { /* earliest timeout has already expired */ + timeout_ptr->sec = 0; + timeout_ptr->usec = 0; + } + + } else { /* NOTE: backend query mechanism must return an error if not watching any fds and timeout_ptr == NULL to avoid sleeping forever */ + timeout_ptr = NULL; + } + + switch (events->backend_type) { + + case E_BACKEND_SELECT: + ret = E_events_select_query(events, query_lvl, timeout_ptr, &queried_lvl); + break; + + case E_BACKEND_EPOLL: + ret = E_events_epoll_query(events, query_lvl, timeout_ptr, &queried_lvl); + break; + + default: + assert(0); + ret = E_BUG; + break; + } + + } while (ret == E_INTERRUPTED); /* loop while interrupted */ + + if (ret != 0) { + goto FAIL; + } + + assert(queried_lvl <= events->max_priority && queried_lvl >= events->active_threshold && queried_lvl <= query_lvl); + + /* generate timeout notices and trim timeouts queue */ + + E_events_update_loop_time(events); /* update events->last_loop_time */ + stdskl_upperb(&events->timeouts, &qit, &events->last_loop_time); /* search for end of tripped timeouts */ + + for (stdskl_begin(&events->timeouts, &sit); !stdskl_it_eq(&sit, &qit); stdskl_it_next(&sit)) { + + watch = *(E_watch**) stdskl_it_val(&sit); + + assert(!watch->noticed); + + if (stddll_push_back(&events->priorities[watch->priority].notices, &watch)) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + watch->noticed = STDTRUE; + stddll_last(&events->priorities[watch->priority].notices, &watch->notice_it); + + ((E_time_watch*) watch)->in_timeouts = STDFALSE; /* removed from timeouts immediately below */ + } + + stdskl_erase_seq(&events->timeouts, stdskl_begin(&events->timeouts, &sit), &qit); + + /* reset starvation counters on all queried levels that are still empty after the query */ + + for (tmp_lvl = events->max_priority; tmp_lvl >= queried_lvl; --tmp_lvl) { + + if (events->priorities[tmp_lvl].starve_thresh != 0 && /* starvation protection enabled */ + events->priorities[tmp_lvl].starve_cnter >= events->priorities[tmp_lvl].starve_thresh && /* starvation threshold reached */ + stddll_empty(&events->priorities[tmp_lvl].notices)) { /* no new notices from query */ + + events->priorities[tmp_lvl].starve_cnter = 0; /* reset the starvation counter */ + } + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0); + + END: + return ret; } -int E_queue( void (* func)( int code, void *data ), int code, void *data, - sp_time delta_time ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_handle_starvation(E_events * events) { - time_event *t_pre; - time_event *t_post; - time_event *t_e; - int inserted; - int deleted; - int compare; + E_priority starve_lvl; + int ret = 0; - t_e = new( TIME_EVENT ); + /* iterate over all active events->priorities */ - t_e->t = E_add_time( E_get_time(), delta_time ); - t_e->func = func; - t_e->code = code; - t_e->data = data; - deleted = 0; - inserted = 0; + for (starve_lvl = events->max_priority; events->keep_running && starve_lvl >= events->active_threshold; --starve_lvl) { + + if (events->priorities[starve_lvl].starve_thresh != 0 && /* starvation protection enabled */ + events->priorities[starve_lvl].starve_cnter >= events->priorities[starve_lvl].starve_thresh) { /* starvation threshold reached */ + + if (!stddll_empty(&events->priorities[starve_lvl].notices)) { /* dispatch starving priority */ + + if ((ret = E_events_dispatch_notice(events, starve_lvl)) != 0) { /* resets starvation counter */ + break; + } + + } else { + ret = E_REPOPULATE; /* this notice queue is empty: we need to try to populate this priority */ + break; + } + } + } - if( Time_queue != NULL ) - { - if( Time_queue->func == t_e->func && - Time_queue->data == t_e->data && - Time_queue->code == t_e->code ) - { - t_pre = Time_queue; - Time_queue = Time_queue->next; - dispose( t_pre ); - deleted = 1; - Alarm( EVENTS, "E_queue: dequeued a (first) simillar event\n" ); - } - } - if( Time_queue == NULL ) - { - t_e->next = NULL; - Time_queue = t_e; - Alarm( EVENTS, "E_queue: (only) event queued func 0x%x code %d data 0x%x in future (%u:%u)\n",t_e->func,t_e->code, t_e->data, delta_time.sec, delta_time.usec ); - return( 0 ); - }else{ - compare = E_compare_time ( t_e->t, Time_queue->t ); - if( compare < 0 ) - { - t_e->next = Time_queue; - Time_queue = t_e; - inserted = 1; - Alarm( EVENTS, "E_queue: (first) event queued func 0x%x code %d data 0x%x in future (%u:%u)\n",t_e->func,t_e->code, t_e->data, delta_time.sec,delta_time.usec ); - } - } - t_pre = Time_queue ; - t_post = Time_queue->next; - while ( t_post != NULL && ( !inserted || !deleted ) ) - { - if( t_post->func == t_e->func && - t_post->data == t_e->data && - t_post->code == t_e->code ) - { - t_pre->next = t_post->next; - dispose( t_post ); - t_post = t_pre->next; - deleted = 1; - Alarm( EVENTS, "E_queue: dequeued a simillar event\n" ); - continue; - } + return ret; +} - if ( !inserted ) - { - compare = E_compare_time ( t_e->t, t_post->t ); - if( compare < 0 ) - { - t_pre->next = t_e; - t_e->next = t_post; - inserted = 1; - Alarm( EVENTS, "E_queue: event queued for func 0x%x code %d data 0x%x in future (%u:%u)\n",t_e->func,t_e->code, t_e->data, delta_time.sec, delta_time.usec ); - } - } +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - t_pre = t_post; - t_post = t_post->next; - } +static int E_events_dispatch_notice(E_events * events, + E_priority dispatch_lvl) +{ + int ret = 0; + E_watch watch_cpy; + E_watch * watch; + stdit dit; - if( !inserted ) - { - t_pre->next = t_e; - t_e->next = NULL; - Alarm( EVENTS, "E_queue: (last) event queued func 0x%x code %d data 0x%x in future (%u:%u)\n",t_e->func,t_e->code, t_e->data, delta_time.sec,delta_time.usec ); - } + assert(dispatch_lvl <= events->max_priority && dispatch_lvl >= events->active_threshold && events->active_threshold >= E_MIN_PRIORITY); + assert(!stddll_empty(&events->priorities[dispatch_lvl].notices)); - return( 0 ); + /* pop the head notice off of dispatch_lvl */ + + watch = *(E_watch**) stddll_it_val(stddll_begin(&events->priorities[dispatch_lvl].notices, &dit)); + assert(watch->priority == dispatch_lvl && watch->noticed && stddll_it_eq(&dit, &watch->notice_it)); + + stddll_erase(&events->priorities[dispatch_lvl].notices, &dit); + watch->noticed = STDFALSE; + + events->priorities[dispatch_lvl].starve_cnter = 0; /* reset starvation counter on this level as we are about to "feed" it */ + + /* execute the callback depending on watch type */ + + if (watch->watch_type == READ_FD || watch->watch_type == WRITE_FD || watch->watch_type == EXCEPT_FD) { + + ((E_fd_callback_fcn) watch->callback_fcn)(((E_fd_watch*) watch)->fd, watch->callback_code, watch->callback_data); + + } else { + E_time_watch * t_watch = (E_time_watch*) watch; + + assert(watch->watch_type == E_DELTA_TIMER || watch->watch_type == E_ABSOLUTE_TIMER || watch->watch_type == E_PERIODIC_TIMER); + assert(!t_watch->in_timeouts); + + if (watch->watch_type != E_PERIODIC_TIMER) { + + watch_cpy = *watch; /* copy so cb params remain valid */ + E_events_cancel_watch(events, watch); /* one shot watch: remove it */ + watch = &watch_cpy; + + } else { + /* TODO: should we use E_events_get_time(), + events->last_clock_time or events->last_loop_time or + something even more complicated that keeps the periodic timer + on schedule according to its periodicity (i.e. - when the + next picket should fall based on initial time of scheduling)? + */ + + sp_time wake_time = E_add_time(E_events_get_time(events), t_watch->timeout); + + if (stdskl_insert(&events->timeouts, &t_watch->timeouts_it, &wake_time, &watch, STDFALSE)) { /* periodic watch: reschedule it */ + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + t_watch->in_timeouts = STDTRUE; + } + + ((E_time_callback_fcn) watch->callback_fcn)(watch->callback_code, watch->callback_data); + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0); + + END: + return ret; } -int E_dequeue( void (* func)( int code, void *data ), int code, - void *data ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_create_timer(E_events * events, + sp_time timeout, + E_watch_type timer_type, + E_time_callback_fcn fcn, + int code, + void * data, + E_priority priority) { - time_event *t_pre; - time_event *t_ptr; + E_time_watch * time_watch = NULL; + E_time_watch * tmp_watch = NULL; + sp_time next_timeout = { 0, 0 }; + int ret = 0; - if( Time_queue == NULL ) - { - Alarm( EVENTS, "E_dequeue: no such event\n" ); - return( -1 ); - } + /* validate input */ - if( Time_queue->func == func && - Time_queue->data == data && - Time_queue->code == code ) - { - t_ptr = Time_queue; - Time_queue = Time_queue->next; - dispose( t_ptr ); - Alarm( EVENTS, "E_dequeue: first event dequeued func 0x%x code %d data 0x%x\n",func,code, data); - return( 0 ); - } + if (E_compare_time(next_timeout, timeout) > 0 || /* negative timeouts not allowed */ + priority > events->max_priority || + priority < E_MIN_PRIORITY) { + ret = E_INVALID_PARAM; + goto FAIL; + } - t_pre = Time_queue; - while ( t_pre->next != NULL ) - { - t_ptr = t_pre->next; - if( t_ptr->func == func && - t_ptr->data == data && - t_ptr->code == code ) - { - t_pre->next = t_ptr->next; - dispose( t_ptr ); - Alarm( EVENTS, "E_dequeue: event dequeued func 0x%x code %d data 0x%x\n",func,code, data); - return( 0 ); - } - t_pre = t_ptr; - } + /* figure out first expiration for this timer */ - Alarm( EVENTS, "E_dequeue: no such event\n" ); - return( -1 ); + switch (timer_type) { + + case E_DELTA_TIMER: + case E_PERIODIC_TIMER: + /* TODO: should we use E_events_get_time(), events->last_clock_time or events->last_loop_time? */ + next_timeout = E_add_time(E_events_get_time(events), timeout); + break; + + case E_ABSOLUTE_TIMER: + /* TODO: if we start using a monotonic timer how does that relate to the wall clock? */ + next_timeout = timeout; + break; + + default: + assert(0); + ret = E_BUG; /* NOTE: currently only internal code calls this fcn and it should call with a valid timer type */ + goto FAIL; + } + + /* allocate and initialize timer */ + + if ((time_watch = (E_time_watch*) calloc(1, sizeof(E_time_watch))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + time_watch->w.watch_type = timer_type; + time_watch->w.priority = priority; + time_watch->w.callback_fcn = fcn; + time_watch->w.callback_code = code; + time_watch->w.callback_data = data; + time_watch->w.noticed = STDFALSE; + time_watch->timeout = timeout; + time_watch->in_timeouts = STDFALSE; /* updated below */ + + /* look up any pre-existing timer with same parameters and insertion spot for new timer */ + + if (!stdskl_is_end(&events->time_watches, stdskl_lowerb(&events->time_watches, &time_watch->time_watches_it, &time_watch))) { + + tmp_watch = *(E_time_watch**) stdskl_it_key(&time_watch->time_watches_it); + + /* if a pre-existing watch exists: cancel old watch */ + + if (E_time_watch_ptr_cmp(&time_watch, &tmp_watch) == 0) { + stdskl_it_next(&time_watch->time_watches_it); /* NOTE: advance so iterator remains valid after cancellation */ + E_events_cancel_watch(events, (E_watch*) tmp_watch); + } + } + + /* insert into main timer lookup table using events->time_watches_it as a hint */ + + if (stdskl_insert(&events->time_watches, &time_watch->time_watches_it, &time_watch, NULL, STDTRUE)) { + ret = E_ALLOC_FAILURE; + goto FAIL_TIME_WATCH; + } + + /* insert into events->timeouts */ + + if (stdskl_insert(&events->timeouts, &time_watch->timeouts_it, &next_timeout, &time_watch, STDFALSE)) { + ret = E_ALLOC_FAILURE; + goto FAIL_TIME_WATCHES; + } + + time_watch->in_timeouts = STDTRUE; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL_TIME_WATCHES: + stdskl_erase(&events->time_watches, &time_watch->time_watches_it); + + FAIL_TIME_WATCH: + free(time_watch); + + FAIL: + assert(ret != 0); + + END: + return ret; } -void E_delay( sp_time t ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_attach_fd(E_events * events, + int fd, + E_watch_type fd_type, + E_fd_callback_fcn fcn, + int code, + void * data, + int priority) { - struct timeval tmp_t; + E_fd_watch * fd_watch; + E_fd_watch * tmp_watch; + int ret = 0; - tmp_t.tv_sec = t.sec; - tmp_t.tv_usec = t.usec; + /* validate input */ -#ifndef ARCH_PC_WIN95 - if (select(0, NULL, NULL, NULL, &tmp_t ) < 0) - { - Alarm( EVENTS, "E_delay: select delay returned error: %s\n", strerror(errno)); - } -#else /* ARCH_PC_WIN95 */ - SleepEx( tmp_t.tv_sec*1000+tmp_t.tv_usec/1000, 0 ); -#endif /* ARCH_PC_WIN95 */ + if (fd < 0 || + priority > events->max_priority || + priority < E_MIN_PRIORITY || + fd < 0) { + ret = E_INVALID_PARAM; + goto FAIL; + } + switch (fd_type) { + + case READ_FD: + case WRITE_FD: + case EXCEPT_FD: + break; + + default: + ret = E_INVALID_PARAM; + goto FAIL; + } + + /* allocate and initialize new fd_watch */ + + if ((fd_watch = (E_fd_watch*) calloc(1, sizeof(E_fd_watch))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + fd_watch->w.watch_type = fd_type; + fd_watch->w.priority = priority; + fd_watch->w.callback_fcn = fcn; + fd_watch->w.callback_code = code; + fd_watch->w.callback_data = data; + fd_watch->w.noticed = STDFALSE; + fd_watch->fd = fd; + fd_watch->active = STDFALSE; /* updated below */ + + /* find insertion point and cancel any previously existing watch with same parameters */ + + if ((tmp_watch = E_events_get_fd_watch(events, fd, fd_type, &fd_watch->fd_watches_it)) != NULL) { + stdskl_it_next(&fd_watch->fd_watches_it); /* NOTE: advance iterator here so it remains valid after cancellation */ + E_events_cancel_watch(events, (E_watch*) tmp_watch); + } + + /* insert into events->fd_watches using events->fd_watches_it as a hint */ + + if (stdskl_insert(&events->fd_watches, &fd_watch->fd_watches_it, &fd, &fd_watch, STDTRUE) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL_FD_WATCH; + } + + /* insert into backend specific monitoring mechanism */ + + switch (events->backend_type) { + + case E_BACKEND_SELECT: + ret = E_events_select_activate_watch(events, fd_watch); + break; + + case E_BACKEND_EPOLL: + ret = E_events_epoll_activate_watch(events, fd_watch); + break; + + default: + assert(0); + ret = E_BUG; + break; + } + + if (ret != 0) { + goto FAIL_FD_WATCHES; + } + + fd_watch->active = STDTRUE; + ++events->priorities[priority].num_active_fd_watches; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL_FD_WATCHES: + stdskl_erase(&events->fd_watches, &fd_watch->fd_watches_it); + + FAIL_FD_WATCH: + free(fd_watch); + + FAIL: + assert(ret != 0); + + END: + return ret; } - -int E_attach_fd( int fd, int fd_type, - void (* func)( mailbox mbox, int code, void *data ), - int code, void *data, int priority ) + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static void E_events_cancel_watch(E_events * events, + E_watch * watch) { - int num_fds; - int j; + /* remove any notice on this watch */ - if( priority < 0 || priority > NUM_PRIORITY ) - { - Alarm( PRINT, "E_attach_fd: invalid priority %d for fd %d with fd_type %d\n", priority, fd, fd_type ); - return( -1 ); - } - if( fd_type < 0 || fd_type > NUM_FDTYPES ) - { - Alarm( PRINT, "E_attach_fd: invalid fd_type %d for fd %d with priority %d\n", fd_type, fd, priority ); - return( -1 ); - } -#ifndef ARCH_PC_WIN95 - /* Windows bug: Reports FD_SETSIZE of 64 but select works on all - * fd's even ones with numbers greater then 64. - */ - if( fd < 0 || fd > FD_SETSIZE ) - { - Alarm( PRINT, "E_attach_fd: invalid fd %d (max %d) with fd_type %d with priority %d\n", fd, FD_SETSIZE, fd_type, priority ); - return( -1 ); - } -#endif - for( j=0; j < Fd_queue[priority].num_fds; j++ ) - { - if( ( Fd_queue[priority].events[j].fd == fd ) && ( Fd_queue[priority].events[j].fd_type == fd_type ) ) - { - Fd_queue[priority].events[j].func = func; - Fd_queue[priority].events[j].code = code; - Fd_queue[priority].events[j].data = data; - if ( !(Fd_queue[priority].events[j].active) ) - Fd_queue[priority].num_active_fds++; - Fd_queue[priority].events[j].active = TRUE; - Alarm( PRINT, - "E_attach_fd: fd %d with type %d exists & replaced & activated\n", fd, fd_type ); - return( 1 ); - } - } - num_fds = Fd_queue[priority].num_fds; + if (watch->noticed) { + stddll_erase(&events->priorities[watch->priority].notices, &watch->notice_it); + watch->noticed = STDFALSE; + } - if ( num_fds == MAX_FD_EVENTS ) { - Alarm( PRINT, "E_attach_fd: Reached Maximum number of events. Recompile with larger MAX_FD_EVENTS\n"); - return( -1 ); - } - Fd_queue[priority].events[num_fds].fd = fd; - Fd_queue[priority].events[num_fds].fd_type = fd_type; - Fd_queue[priority].events[num_fds].func = func; - Fd_queue[priority].events[num_fds].code = code; - Fd_queue[priority].events[num_fds].data = data; - Fd_queue[priority].events[num_fds].active = TRUE; - Fd_queue[priority].num_fds++; - Fd_queue[priority].num_active_fds++; - if( Active_priority <= priority ) FD_SET( fd, &Fd_mask[fd_type] ); + /* remove the watch from its containers based on its type */ - Alarm( EVENTS, "E_attach_fd: fd %d, fd_type %d, code %d, data 0x%x, priority %d Active_priority %d\n", - fd, fd_type, code, data, priority, Active_priority ); + switch (watch->watch_type) { - return( 0 ); + case READ_FD: + case WRITE_FD: + case EXCEPT_FD: { + E_fd_watch * fd_watch = (E_fd_watch*) watch; + + stdskl_erase(&events->fd_watches, &fd_watch->fd_watches_it); + + if (fd_watch->active) { + + switch (events->backend_type) { + + case E_BACKEND_SELECT: + E_events_select_deactivate_watch(events, fd_watch); + break; + + case E_BACKEND_EPOLL: + E_events_epoll_deactivate_watch(events, fd_watch); + break; + + default: + assert(0); + break; + } + + fd_watch->active = STDFALSE; + --events->priorities[watch->priority].num_active_fd_watches; + } + + break; + } + + case E_DELTA_TIMER: + case E_ABSOLUTE_TIMER: + case E_PERIODIC_TIMER: { + E_time_watch * time_watch = (E_time_watch*) watch; + + stdskl_erase(&events->time_watches, &time_watch->time_watches_it); + + if (time_watch->in_timeouts) { + stdskl_erase(&events->timeouts, &time_watch->timeouts_it); + time_watch->in_timeouts = STDFALSE; + } + + break; + } + + default: + assert(0); + break; + } + + /* deallocate the watch */ + + free(watch); } -int E_detach_fd( int fd, int fd_type ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_queue(E_events * events, + E_time_callback_fcn fcn, + int code, + void * data, + sp_time delta_time) { - int i,j; - int found; + return E_events_create_timer(events, delta_time, E_DELTA_TIMER, fcn, code, data, HIGH_PRIORITY); +} - if( fd_type < 0 || fd_type > NUM_FDTYPES ) - { - Alarm( PRINT, "E_detach_fd: invalid fd_type %d for fd %d\n", fd_type, fd ); - return( -1 ); - } +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - found = 0; - for( i=0; i < NUM_PRIORITY; i++ ) - for( j=0; j < Fd_queue[i].num_fds; j++ ) - { - if( ( Fd_queue[i].events[j].fd == fd ) && ( Fd_queue[i].events[j].fd_type == fd_type ) ) - { - if (Fd_queue[i].events[j].active) - Fd_queue[i].num_active_fds--; - Fd_queue[i].num_fds--; - Fd_queue[i].events[j] = Fd_queue[i].events[Fd_queue[i].num_fds]; +static int E_events_queue_absolute(E_events * events, + E_time_callback_fcn fcn, + int code, + void * data, + sp_time absolute_time) +{ + return E_events_create_timer(events, absolute_time, E_ABSOLUTE_TIMER, fcn, code, data, HIGH_PRIORITY); +} - FD_CLR( fd, &Fd_mask[fd_type] ); - found = 1; +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - break; /* from the j for only */ - } - } +static int E_events_queue_periodic(E_events * events, + E_time_callback_fcn fcn, + int code, + void * data, + sp_time periodic_time) +{ + return E_events_create_timer(events, periodic_time, E_PERIODIC_TIMER, fcn, code, data, HIGH_PRIORITY); +} - if( ! found ) return( -1 ); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - return( 0 ); +static int E_events_dequeue(E_events * events, + E_time_callback_fcn fcn, + int code, + void * data) +{ + E_time_watch time_watch_impl; + E_time_watch * time_watch = &time_watch_impl; + stdit sit; + int ret = -1; /* backwards compatability */ + + /* look up any existing time watch and cancel it */ + + time_watch->w.callback_fcn = fcn; + time_watch->w.callback_code = code; + time_watch->w.callback_data = data; + + if (!stdskl_is_end(&events->time_watches, stdskl_find(&events->time_watches, &sit, &time_watch))) { + E_events_cancel_watch(events, *(E_watch**) stdskl_it_key(&sit)); + ret = 0; + } + + return ret; } -int E_deactivate_fd( int fd, int fd_type ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_detach_fd(E_events * events, + int fd, + E_watch_type fd_type) { - int i,j; - int found; + E_fd_watch * fd_watch; + stdit sit; + int ret = -1; /* backwards compatability */ - if( fd_type < 0 || fd_type > NUM_FDTYPES ) - { - Alarm( PRINT, "E_deactivate_fd: invalid fd_type %d for fd %d\n", fd_type, fd ); - return( -1 ); - } + /* cancel any previously existing watch */ - found = 0; + if ((fd_watch = E_events_get_fd_watch(events, fd, fd_type, &sit)) != NULL) { + ret = 0; + E_events_cancel_watch(events, (E_watch*) fd_watch); + } - for( i=0; i < NUM_PRIORITY; i++ ) - for( j=0; j < Fd_queue[i].num_fds; j++ ) - { - if( ( Fd_queue[i].events[j].fd == fd ) && ( Fd_queue[i].events[j].fd_type == fd_type ) ) - { - if (Fd_queue[i].events[j].active) - Fd_queue[i].num_active_fds--; - Fd_queue[i].events[j].active = FALSE; - FD_CLR( fd, &Fd_mask[fd_type] ); - found = 1; + return ret; +} - break; /* from the j for only */ - } - } +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - if( ! found ) return( -1 ); - return( 0 ); +static int E_events_set_active_threshold(E_events * events, + E_priority priority) +{ + if (priority < E_MIN_PRIORITY || priority > events->max_priority) { + return -1; /* backwards compatability */ + } + + events->active_threshold = priority; + + return priority; /* backwards compatability */ } -int E_activate_fd( int fd, int fd_type ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_num_active(E_events * events, + E_priority priority) { - int i,j; - int found; + if (priority < E_MIN_PRIORITY || priority > events->max_priority) { + return -1; /* backwards compatability */ + } - if( fd_type < 0 || fd_type > NUM_FDTYPES ) - { - Alarm( PRINT, "E_activate_fd: invalid fd_type %d for fd %d\n", fd_type, fd ); - return( -1 ); - } + return events->priorities[priority].num_active_fd_watches; +} - found = 0; +/*************************************************************************************************************************** + ***************************************************************************************************************************/ - for( i=0; i < NUM_PRIORITY; i++ ) - for( j=0; j < Fd_queue[i].num_fds; j++ ) - { - if( ( Fd_queue[i].events[j].fd == fd ) && ( Fd_queue[i].events[j].fd_type == fd_type ) ) - { - if ( !(Fd_queue[i].events[j].active) ) - Fd_queue[i].num_active_fds++; - Fd_queue[i].events[j].active = TRUE; - if( i >= Active_priority ) FD_SET( fd, &Fd_mask[ fd_type ] ); - found = 1; +static int E_events_activate_fd(E_events * events, + int fd, + E_watch_type fd_type) +{ + E_fd_watch * fd_watch; + stdit sit; + int ret = -1; /* backwards compatability */ - break; /* from the j for only */ - } - } + if ((fd_watch = E_events_get_fd_watch(events, fd, fd_type, &sit)) != NULL) { - if( ! found ) return( -1 ); - return( 0 ); + ret = 0; + + if (!fd_watch->active) { + + switch (events->backend_type) { + + case E_BACKEND_SELECT: + ret = E_events_select_activate_watch(events, fd_watch); + break; + + case E_BACKEND_EPOLL: + ret = E_events_epoll_activate_watch(events, fd_watch); + break; + + default: + assert(0); + ret = E_BUG; + break; + } + + if (ret == 0) { + fd_watch->active = STDTRUE; + ++events->priorities[fd_watch->w.priority].num_active_fd_watches; + } + } + } + + return ret; } -int E_set_active_threshold( int priority ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_deactivate_fd(E_events * events, + int fd, + E_watch_type fd_type) { - int fd_type; - int i,j; + E_fd_watch * fd_watch; + stdit sit; + int ret = -1; /* backwards compatability */ - if( priority < 0 || priority > NUM_PRIORITY ) - { - Alarm( PRINT, "E_set_active_threshold: invalid priority %d\n", priority ); - return( -1 ); - } + if ((fd_watch = E_events_get_fd_watch(events, fd, fd_type, &sit)) != NULL) { - if( priority == Active_priority ) return( priority ); + ret = 0; - Active_priority = priority; - for ( i=0; i < NUM_FDTYPES; i++ ) - { - FD_ZERO( &Fd_mask[i] ); - } + if (fd_watch->active) { + + switch (events->backend_type) { + + case E_BACKEND_SELECT: + E_events_select_deactivate_watch(events, fd_watch); /* TODO: should we look at return value or not? */ + break; + + case E_BACKEND_EPOLL: + E_events_epoll_deactivate_watch(events, fd_watch); /* TODO: should we look at return value or not? */ + break; + + default: + assert(0); + ret = E_BUG; + break; + } - for( i = priority; i < NUM_PRIORITY; i++ ) - for( j=0; j < Fd_queue[i].num_fds; j++ ) - { - fd_type = Fd_queue[i].events[j].fd_type; - if (Fd_queue[i].events[j].active) - FD_SET( Fd_queue[i].events[j].fd, &Fd_mask[fd_type] ); - } + --events->priorities[fd_watch->w.priority].num_active_fd_watches; + fd_watch->active = STDFALSE; + } + } - Alarm( EVENTS, "E_set_active_threshold: changed to %d\n",Active_priority); + return ret; +} - return( priority ); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_set_elevate_count(E_events * events, + E_priority priority, + unsigned count) +{ + if (priority < E_MIN_PRIORITY || priority > events->max_priority) { + return E_INVALID_PARAM; + } + + if ((events->priorities[priority].starve_thresh = count) == 0) { + events->priorities[priority].starve_cnter = 0; + } + + return 0; } -int E_num_active( int priority ) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_get_elevate_count(E_events * events, + E_priority priority, + unsigned * count) { - if( priority < 0 || priority > NUM_PRIORITY ) - { - Alarm( PRINT, "E_num_active: invalid priority %d\n", priority ); - return( -1 ); - } - return( Fd_queue[priority].num_active_fds ); + if (priority < E_MIN_PRIORITY || priority > events->max_priority) { + return E_INVALID_PARAM; + } + + *count = events->priorities[priority].starve_thresh; + + return 0; } -void E_handle_events(void) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_select_init(E_events * events) { -static int Round_robin = 0; -static const sp_time long_timeout = { 10000, 0}; -static const sp_time zero_sec = { 0, 0}; -#ifdef BADCLOCK -static const sp_time mili_sec = { 0, 1000}; - int clock_sync; -#endif - int num_set; - int treated; - int fd; - int fd_type; - int i,j; - sp_time timeout; - struct timeval sel_timeout, wait_timeout; - fd_set current_mask[NUM_FDTYPES]; - time_event *temp_ptr; - int first=1; -#ifdef TESTTIME - sp_time tmp_late,start,stop,req_time; /* DEBUGGING */ -#endif -#ifdef BADCLOCK - clock_sync = 0; -#endif - for( Exit_events = 0 ; !Exit_events ; ) - { - Alarm( EVENTS, "E_handle_events: next event \n"); + int priority; + E_select_info * info; + E_select_lvl_info * lvl_info; + E_watch_type fd_type; + int ret = 0; - /* Handle time events */ - timeout = long_timeout; -#ifdef TESTTIME - start = E_get_time(); -#endif - while( Time_queue != NULL ) - { -#ifdef BADCLOCK - if ( clock_sync >= 0 ) - { - E_get_time(); - clock_sync = -20; - } + assert(events->backend_type == E_BACKEND_UNSPECIFIED); + + if ((info = (E_select_info*) calloc(1, sizeof(E_select_info))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + for (priority = E_MIN_PRIORITY; priority != events->max_priority; ++priority) { + + if ((lvl_info = (E_select_lvl_info*) calloc(1, sizeof(E_select_lvl_info))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL_PRIORITY; + } + + for (fd_type = 0; fd_type != NUM_FDTYPES; ++fd_type) { + + FD_ZERO(&lvl_info->fd_sets[fd_type]); + + if (stdskl_construct(&lvl_info->fd_skls[fd_type], sizeof(int), 0, E_int_cmp) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL_LVL_INFO; + } + } + + events->priorities[priority].backend_lvl_info = lvl_info; + } + + events->backend_type = E_BACKEND_SELECT; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL_LVL_INFO: + while (fd_type-- != 0) { + stdskl_destruct(&lvl_info->fd_skls[fd_type]); + } + + free(lvl_info); + + FAIL_PRIORITY: + while (priority-- != E_MIN_PRIORITY) { + + lvl_info = (E_select_lvl_info*) events->priorities[priority].backend_lvl_info; + + for (fd_type = 0; fd_type != NUM_FDTYPES; ++fd_type) { + stdskl_destruct(&lvl_info->fd_skls[fd_type]); + } + + free(lvl_info); + events->priorities[priority].backend_lvl_info = NULL; + } + + free(info); + + FAIL: + info = NULL; + assert(ret != 0); + + END: + events->backend_info = info; + + return ret; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static void E_events_select_fini(E_events * events) +{ + E_priority priority; + E_select_info * info; + E_select_lvl_info * lvl_info; + E_watch_type fd_type; + + assert(events->backend_type == E_BACKEND_SELECT && events->backend_info != NULL); + + for (priority = E_MIN_PRIORITY; priority != events->max_priority; ++priority) { + + lvl_info = (E_select_lvl_info*) events->priorities[priority].backend_lvl_info; + assert(lvl_info != NULL); + + for (fd_type = 0; fd_type != NUM_FDTYPES; ++fd_type) { + stdskl_destruct(&lvl_info->fd_skls[fd_type]); + } + + free(lvl_info); + events->priorities[priority].backend_lvl_info = NULL; + } + + info = (E_select_info*) events->backend_info; + free(info); + + events->backend_info = NULL; + events->backend_type = E_BACKEND_UNSPECIFIED; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_select_activate_watch(E_events * events, + E_fd_watch * fd_watch) +{ + E_priority lvl; + E_select_lvl_info * lvl_info; + stdit sit; + int ret = 0; + + assert(fd_watch == E_events_get_fd_watch(events, fd_watch->fd, fd_watch->w.watch_type, &sit)); + assert(fd_watch->w.priority >= E_MIN_PRIORITY && fd_watch->w.priority <= events->max_priority); + assert(fd_watch->w.watch_type >= 0 && fd_watch->w.watch_type <= NUM_FDTYPES); + assert(!fd_watch->active); + + /* go down to the watch's priority lvl adding the new watch to each priority's fd sets */ + + for (lvl = events->max_priority; lvl >= fd_watch->w.priority; --lvl) { + + lvl_info = (E_select_lvl_info*) events->priorities[fd_watch->w.priority].backend_lvl_info; + + assert(!FD_ISSET(fd_watch->fd, &lvl_info->fd_sets[fd_watch->w.watch_type])); + assert(!stdskl_contains(&lvl_info->fd_skls[fd_watch->w.watch_type], &fd_watch->fd)); + + if (stdskl_put(&lvl_info->fd_skls[fd_watch->w.watch_type], &sit, &fd_watch->fd, NULL, STDFALSE) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + FD_SET(fd_watch->fd, &lvl_info->fd_sets[fd_watch->w.watch_type]); + } + + fd_watch->active = STDTRUE; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + E_events_select_deactivate_watch(events, fd_watch); + assert(ret != 0); + + END: + return ret; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_select_deactivate_watch(E_events * events, + E_fd_watch * fd_watch) +{ + E_priority lvl; + E_select_lvl_info * lvl_info; + stdit sit; + + assert(fd_watch == E_events_get_fd_watch(events, fd_watch->fd, fd_watch->w.watch_type, &sit)); + assert(fd_watch->w.priority >= E_MIN_PRIORITY && fd_watch->w.priority <= events->max_priority); + assert(fd_watch->w.watch_type >= 0 && fd_watch->w.watch_type <= NUM_FDTYPES); + + /* NOTE: we can't assert(fd_watch->active) bc we use this fcn to + clean up errors in activate_watch(). That's also the reason why + our logic must be conditional within the loop. + */ + + /* go down to the watch's priority lvl adding the new watch to each priority's fd sets */ + + for (lvl = events->max_priority; lvl >= fd_watch->w.priority; --lvl) { + + lvl_info = (E_select_lvl_info*) events->priorities[fd_watch->w.priority].backend_lvl_info; + + if (FD_ISSET(fd_watch->fd, &lvl_info->fd_sets[fd_watch->w.watch_type])) { + + assert(stdskl_contains(&lvl_info->fd_skls[fd_watch->w.watch_type], &fd_watch->fd)); + stdskl_erase_key(&lvl_info->fd_skls[fd_watch->w.watch_type], &fd_watch->fd); + FD_CLR(fd_watch->fd, &lvl_info->fd_sets[fd_watch->w.watch_type]); + + } else { + assert(!stdskl_contains(&lvl_info->fd_skls[fd_watch->w.watch_type], &fd_watch->fd)); + } + } + + fd_watch->active = STDFALSE; + + return 0; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_epoll_init(E_events * events) +#ifdef HAVE_LINUX +{ + int priority; + E_epoll_info * info; + E_epoll_lvl_info * lvl_info; + int ret = 0; + + assert(events->backend_type == E_BACKEND_UNSPECIFIED); + assert(E_EPOLL_INITIAL_SIZE > 0); + + if ((info = (E_epoll_info*) calloc(1, sizeof(E_epoll_info))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + if ((info->events = (struct epoll_event*) calloc(E_EPOLL_INITIAL_SIZE, sizeof(struct epoll_event))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL_INFO; + } + + info->num_events = E_EPOLL_INITIAL_SIZE; + + for (priority = E_MIN_PRIORITY; priority != events->max_priority; ++priority) { + + if ((lvl_info = (E_epoll_lvl_info*) calloc(1, sizeof(E_epoll_lvl_info))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL_PRIORITY; + } + + if ((lvl_info->epoll_fd = epoll_create(E_EPOLL_INITIAL_SIZE)) < 0) { + + switch (errno) { + + case EINVAL: + assert(0); + ret = E_BUG; + break; + + case ENFILE: + ret = E_TOO_MANY_FILES; + break; + + case ENOMEM: + ret = E_ALLOC_FAILURE; + break; + + default: + ret = E_SYS_FAILURE; + break; + } + + goto FAIL_LVL_INFO; + } + + lvl_info->num_poll_fds = 0; + events->priorities[priority].backend_lvl_info = lvl_info; + } + + events->backend_type = E_BACKEND_EPOLL; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL_LVL_INFO: + free(lvl_info); + + FAIL_PRIORITY: + while (priority-- != E_MIN_PRIORITY) { + lvl_info = (E_epoll_lvl_info*) events->priorities[priority].backend_lvl_info; + + close(lvl_info->epoll_fd); + free(lvl_info); + + events->priorities[priority].backend_lvl_info = NULL; + } + + free(info->events); + + FAIL_INFO: + free(info); + + FAIL: + info = NULL; + assert(ret != 0); + + END: + events->backend_info = info; + + return ret; +} #else - E_get_time(); +{ + return E_NO_SYS; +} #endif - if ( !first && E_compare_time( Now, Time_queue->t ) >= 0 ) - { -#ifdef TESTTIME - tmp_late = E_sub_time( Now, Time_queue->t ); -#endif - temp_ptr = Time_queue; - Time_queue = Time_queue->next; - Alarm( EVENTS, "E_handle_events: exec time event \n"); -#ifdef TESTTIME - Alarm( DEBUG, "Events: TimeEv is %d %d late\n",tmp_late.sec, tmp_late.usec); -#endif - temp_ptr->func( temp_ptr->code, temp_ptr->data ); - dispose( temp_ptr ); -#ifdef BADCLOCK - Now = E_add_time( Now, mili_sec ); - clock_sync++; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static void E_events_epoll_fini(E_events * events) +#ifdef HAVE_LINUX +{ + E_priority priority; + E_epoll_info * info; + E_epoll_lvl_info * lvl_info; + + assert(events->backend_type == E_BACKEND_EPOLL && events->backend_info != NULL); + + for (priority = E_MIN_PRIORITY; priority != events->max_priority; ++priority) { + + lvl_info = (E_epoll_lvl_info*) events->priorities[priority].backend_lvl_info; + assert(lvl_info != NULL); + + close(lvl_info->epoll_fd); + free(lvl_info); + + events->priorities[priority].backend_lvl_info = NULL; + } + + info = (E_epoll_info*) events->backend_info; + free(info->events); + free(info); + + events->backend_info = NULL; + events->backend_type = E_BACKEND_UNSPECIFIED; +} #else - E_get_time(); +{ +} #endif - if (Exit_events) goto end_handler; - }else{ - timeout = E_sub_time( Time_queue->t, Now ); - break; - } - } - if (timeout.sec < 0 ) - timeout.sec = timeout.usec = 0; /* this can happen until first is unset */ -#ifdef TESTTIME - stop = E_get_time(); - tmp_late = E_sub_time(stop, start); - Alarm(DEBUG, "Events: TimeEv's took %d %d to handle\n", tmp_late.sec, tmp_late.usec); -#endif - /* Handle fd events */ - for( i=0; i < NUM_FDTYPES; i++ ) - { - current_mask[i] = Fd_mask[i]; - } - Alarm( EVENTS, "E_handle_events: poll select\n"); -#ifdef TESTTIME - req_time = zero_sec; -#endif - wait_timeout.tv_sec = zero_sec.sec; - wait_timeout.tv_usec = zero_sec.usec; - num_set = select( FD_SETSIZE, ¤t_mask[READ_FD], ¤t_mask[WRITE_FD], ¤t_mask[EXCEPT_FD], - &wait_timeout ); - if (num_set == 0 && !Exit_events) - { -#ifdef BADCLOCK - clock_sync = 0; -#endif - for( i=0; i < NUM_FDTYPES; i++ ) - { - current_mask[i] = Fd_mask[i]; - } - Alarm( EVENTS, "E_handle_events: select with timeout (%d, %d)\n", - timeout.sec,timeout.usec ); -#ifdef TESTTIME - req_time = E_add_time(req_time, timeout); -#endif - sel_timeout.tv_sec = timeout.sec; - sel_timeout.tv_usec = timeout.usec; - num_set = select( FD_SETSIZE, ¤t_mask[READ_FD], ¤t_mask[WRITE_FD], - ¤t_mask[EXCEPT_FD], &sel_timeout ); - } -#ifdef TESTTIME - start = E_get_time(); - tmp_late = E_sub_time(start, stop); - Alarm( DEBUG, "Events: Waiting for fd or timout took %d %d asked for %d %d\n", tmp_late.sec, tmp_late.usec, req_time.sec, req_time.usec); -#endif - /* Handle all high and medium priority fd events */ - for( i=NUM_PRIORITY-1,treated=0; - i > LOW_PRIORITY && num_set > 0 && !treated; - i-- ) - { - for( j=0; j < Fd_queue[i].num_fds && num_set > 0; j++ ) - { - fd = Fd_queue[i].events[j].fd; - fd_type = Fd_queue[i].events[j].fd_type; - if( FD_ISSET( fd, ¤t_mask[fd_type] ) ) - { - Alarm( EVENTS, "E_handle_events: exec handler for fd %d, fd_type %d, priority %d\n", - fd, fd_type, i ); - Fd_queue[i].events[j].func( - Fd_queue[i].events[j].fd, - Fd_queue[i].events[j].code, - Fd_queue[i].events[j].data ); - treated = 1; - num_set--; -#ifdef BADCLOCK - Now = E_add_time( Now, mili_sec ); - clock_sync++; + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_epoll_activate_watch(E_events * events, + E_fd_watch * fd_watch) +#ifdef HAVE_LINUX +{ + E_priority lvl; + E_epoll_lvl_info * lvl_info; + E_fd_watch * tmp_watch; + int num_events; + struct epoll_event epoll_evnt; + struct epoll_event epoll_evnt_cpy; + int epoll_op; + stdit fd_it; + stdit sit; + int ret = 0; + + assert(!fd_watch->active); + + epoll_evnt.data.fd = fd_watch->fd; + stdskl_lowerb(&events->fd_watches, &fd_it, &fd_watch->fd); /* find the beginning of all the watches on the fd */ + + /* go down to the watch's priority lvl adding the new watch to each priority's epoll fd sets */ + + for (lvl = events->max_priority; lvl >= fd_watch->w.priority; --lvl) { + + lvl_info = (E_epoll_lvl_info*) events->priorities[fd_watch->w.priority].backend_lvl_info; + + /* calculate what events we should query for at this priority lvl for all the other watches on this fd */ + + epoll_evnt.events = 0; + num_events = 0; + + for (sit = fd_it; !stdskl_is_end(&events->fd_watches, &sit); stdskl_it_next(&sit)) { /* go over all the watches on this fd */ + + tmp_watch = *(E_fd_watch**) stdskl_it_val(&sit); + + if (tmp_watch->fd != fd_watch->fd) { /* end of watches on this fd */ + break; + } + + if (!tmp_watch->active || tmp_watch == fd_watch) { /* skip inactive watches (we handle fd_watch after this loop) */ + continue; + } + + if (tmp_watch->w.priority >= lvl) { /* is this watch participating in this lvl? */ + + switch (tmp_watch->w.watch_type) { /* how is this watch participating? */ + + case READ_FD: + epoll_evnt.events |= EPOLLIN; + break; + + case WRITE_FD: + epoll_evnt.events |= EPOLLOUT; + break; + + case EXCEPT_FD: + epoll_evnt.events |= EPOLLPRI; + break; + + default: + ret = E_BUG; + assert(0); + goto FAIL; + } + + ++num_events; + } + } + + /* now include fd_watch */ + + switch (fd_watch->w.watch_type) { /* how is this watch participating? */ + + case READ_FD: + epoll_evnt.events |= EPOLLIN; + break; + + case WRITE_FD: + epoll_evnt.events |= EPOLLOUT; + break; + + case EXCEPT_FD: + epoll_evnt.events |= EPOLLPRI; + break; + + default: + assert(0); + ret = E_BUG; + goto FAIL; + } + + ++num_events; + + /* now actually modify the epoll set with the calculated events */ + + epoll_op = (num_events == 1 ? EPOLL_CTL_ADD : EPOLL_CTL_MOD); + + RETRY: /* allow a single retry w/ a different epoll_op in the case of ENOENT (see below) */ + epoll_evnt_cpy = epoll_evnt; /* epoll_ctl's interface isn't const so make a copy in case they trash our input */ + + if ((ret = epoll_ctl(lvl_info->epoll_fd, epoll_op, fd_watch->fd, &epoll_evnt_cpy)) != 0) { + + ret = errno; + assert(ret != 0); + + switch (ret) { + + case ENOENT: /* can happen if watched fd's were close()'d, which epoll notices, without the user calling E_detach_fd() */ + assert(epoll_op == EPOLL_CTL_MOD); + epoll_op = EPOLL_CTL_ADD; /* call again with ADD instead of MOD */ + goto RETRY; + + case EEXIST: /* bug either in epoll_ctl or in our code: shouldn't happen */ + assert(epoll_op == EPOLL_CTL_ADD); + assert(0); + ret = E_BUG; + goto FAIL; + + case EBADF: /* bad fd from user */ + case EPERM: + ret = E_INVALID_PARAM; + goto FAIL; + + case EINVAL: + assert(0); + ret = E_BUG; + goto FAIL; + + case ENOMEM: + default: + ret = E_SYS_FAILURE; + goto FAIL; + } + } + + if (num_events == 1) { + assert(lvl_info->num_poll_fds >= 0); + ++lvl_info->num_poll_fds; + assert(lvl_info->num_poll_fds >= 0); + } + } + + fd_watch->active = STDTRUE; + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + E_events_epoll_deactivate_watch(events, fd_watch); + assert(ret != 0); + + END: + return ret; +} #else - E_get_time(); +{ + return E_NO_SYS; +} #endif - if (Exit_events) goto end_handler; - } - } - } - /* Don't handle timed events until all non-low-priority fd events have been handled - * FIXME: This may or may not be right. If continual high priority events occur, then - * timed events will starve, I'm not sure if that is better then what we have. We - * could also set first=0 no matter what after trying the high events once, then - * they will get a shot first, but after that timed events will also be handled. - */ - if (!treated) - first = 0; -#ifdef TESTTIME - stop = E_get_time(); - tmp_late = E_sub_time(stop, start); - Alarm(DEBUG, "Events: High & Med took %d %d time to handle\n", tmp_late.sec, tmp_late.usec); +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_epoll_deactivate_watch(E_events * events, + E_fd_watch * fd_watch) +#ifdef HAVE_LINUX +{ + E_priority lvl; + E_epoll_lvl_info * lvl_info; + E_fd_watch * tmp_watch; + int num_events; + struct epoll_event epoll_evnt; + int epoll_op; + stdit fd_it; + stdit sit; + int ret = 0; + + epoll_evnt.data.fd = fd_watch->fd; + stdskl_lowerb(&events->fd_watches, &fd_it, &fd_watch->fd); /* find the beginning of all the watches on the fd */ + + /* go down to the watch's priority lvl removing the watch from each priority's epoll fd sets */ + + for (lvl = events->max_priority; lvl >= fd_watch->w.priority; --lvl) { + + lvl_info = (E_epoll_lvl_info*) events->priorities[fd_watch->w.priority].backend_lvl_info; + + /* calculate what events we should query for at this priority lvl for all the other watches on this fd */ + + epoll_evnt.events = 0; + num_events = 0; + + for (sit = fd_it; !stdskl_is_end(&events->fd_watches, &sit); stdskl_it_next(&sit)) { /* go over all the watches on this fd */ + + tmp_watch = *(E_fd_watch**) stdskl_it_val(&sit); + + if (tmp_watch->fd != fd_watch->fd) { /* end of watches on this fd */ + break; + } + + if (!tmp_watch->active || tmp_watch == fd_watch) { /* skip inactive watches and fd_watch */ + continue; + } + + if (tmp_watch->w.priority >= lvl) { /* is this watch participating in this lvl? */ + + switch (tmp_watch->w.watch_type) { /* how is this watch participating? */ + + case READ_FD: + epoll_evnt.events |= EPOLLIN; + break; + + case WRITE_FD: + epoll_evnt.events |= EPOLLOUT; + break; + + case EXCEPT_FD: + epoll_evnt.events |= EPOLLPRI; + break; + + default: + assert(0); + ret = E_BUG; + goto FAIL; + } + + ++num_events; + } + } + + /* now actually modify the epoll set with the calculated events */ + + epoll_op = (num_events == 0 ? EPOLL_CTL_DEL : EPOLL_CTL_MOD); + + if ((ret = epoll_ctl(lvl_info->epoll_fd, epoll_op, fd_watch->fd, &epoll_evnt)) != 0) { + + ret = errno; + assert(ret != 0); + + switch (ret) { + + case ENOENT: /* can happen if watched fd's were close()'d, which epoll notices, without the user calling E_detach_fd() */ + case EPERM: /* same as above, but user opened a new fd (w/ same fd number) that can't be polled */ + case EBADF: /* can happen if user closed fd or if somehow closed our epoll fd (which would be a big no-no) */ + case EINVAL: /* shouldn't happen (might happen if user somehow closed our epoll fd, which would be a big no-no, or tried to watch it) */ + case EEXIST: /* bug either in epoll_ctl or in our code: shouldn't happen */ + case ENOMEM: /* shouldn't happen if Linux has any brains */ + default: + ret = 0; /* ignore errors */ + break; + } + } + + if (num_events == 0) { + assert(lvl_info->num_poll_fds >= 0); + --lvl_info->num_poll_fds; + assert(lvl_info->num_poll_fds >= 0); + } + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0); + + END: + return ret; +} +#else +{ + return E_NO_SYS; +} #endif - /* Handle one low priority fd event. - However, verify that Active_priority still allows LOW_PRIORITY events. - Active_priority can change because of calls to E_set_threshold() during the current select loop. - */ - for( i=0; i < Fd_queue[LOW_PRIORITY].num_fds - && num_set > 0 - && Active_priority == LOW_PRIORITY; - i++ ) - { - j = ( i + Round_robin ) % Fd_queue[LOW_PRIORITY].num_fds; - fd = Fd_queue[LOW_PRIORITY].events[j].fd; - fd_type = Fd_queue[LOW_PRIORITY].events[j].fd_type; - if( FD_ISSET( fd, ¤t_mask[fd_type] ) ) - { - Round_robin = ( j + 1 ) % Fd_queue[LOW_PRIORITY].num_fds; - Alarm( EVENTS , "E_handle_events: exec ext fd event \n"); - Fd_queue[LOW_PRIORITY].events[j].func( - Fd_queue[LOW_PRIORITY].events[j].fd, - Fd_queue[LOW_PRIORITY].events[j].code, - Fd_queue[LOW_PRIORITY].events[j].data ); - num_set--; -#ifdef BADCLOCK - Now = E_add_time( Now, mili_sec ); - clock_sync++; +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_epoll_query(E_events * events, + E_priority query_lvl, + const sp_time * delta_timeout, + E_priority * queried_lvl) +#ifdef HAVE_LINUX +{ + E_epoll_info * info = (E_epoll_info*) events->backend_info; + E_epoll_lvl_info * lvl_info = (E_epoll_lvl_info*) events->priorities[query_lvl].backend_lvl_info; + double tmp_timeout; + int timeout; + struct epoll_event * evnt; + struct epoll_event * evnts_end; + E_fd_watch * fd_watch; + stdit sit; + int ret = 0; + + assert(info->num_events >= lvl_info->num_poll_fds); + + /* convert the timeout to epoll's input */ + + if (delta_timeout != NULL) { + + tmp_timeout = delta_timeout->sec * 1000.0 + delta_timeout->usec / 1000.0; + assert(tmp_timeout >= 0.0); + + if (tmp_timeout <= INT_MAX) { + timeout = (int) tmp_timeout; + + } else { + timeout = INT_MAX; + } + + } else if (lvl_info->num_poll_fds != 0) { + timeout = -1; + + } else { + ret = E_DEADEND; /* error on infinite timeout and no fds to watch */ + goto FAIL; + } + + /* ensure our query array is still large enough */ + + if (lvl_info->num_poll_fds > info->num_events) { + + if ((evnt = (struct epoll_event*) calloc(2 * lvl_info->num_poll_fds, sizeof(struct epoll_event))) == NULL) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + info->events = evnt; + info->num_events = 2 * lvl_info->num_poll_fds; + } + + /* query epoll */ + + if ((ret = epoll_wait(lvl_info->epoll_fd, info->events, info->num_events, timeout)) < 0) { + + ret = errno; + assert(ret != 0); + + switch (ret) { + + case EINTR: + ret = E_INTERRUPTED; /* will cause this fcn to be called again */ + goto FAIL; + + case EBADF: + case EINVAL: + case EFAULT: + assert(0); + ret = E_BUG; + goto FAIL; + + default: + ret = E_SYS_FAILURE; + goto FAIL; + } + } + + /* process the result set */ + + for (evnt = info->events, evnts_end = info->events + ret; evnt != evnts_end; ++evnt) { + + /* look up the fd and process all watches on it */ + + for (stdskl_lowerb(&events->fd_watches, &sit, &evnt->data.fd); !stdskl_is_end(&events->fd_watches, &sit); stdskl_it_next(&sit)) { + + stdbool generate_notice = STDFALSE; + + fd_watch = (E_fd_watch*) stdskl_it_val(&sit); + + if (fd_watch->fd != evnt->data.fd) { /* we've moved onto the next fd */ + break; + } + + if (fd_watch->w.noticed) { /* this watch is already noticed */ + continue; + } + + switch (fd_watch->w.watch_type) { + + case READ_FD: + generate_notice = ((evnt->events & (EPOLLIN | EPOLLERR | EPOLLHUP)) != 0); + break; + + case WRITE_FD: + generate_notice = ((evnt->events & (EPOLLOUT | EPOLLERR | EPOLLHUP)) != 0); + break; + + case EXCEPT_FD: + generate_notice = ((evnt->events & (EPOLLPRI | EPOLLERR | EPOLLHUP)) != 0); + break; + } + + /* push a new notice onto the associated priority's notice queue and record where it is */ + + if (generate_notice && + stddll_insert(&events->priorities[fd_watch->w.priority].notices, + stddll_end(&events->priorities[fd_watch->w.priority].notices, &fd_watch->w.notice_it), + &fd_watch) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + fd_watch->w.noticed = STDTRUE; + } + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0); + + END: + return ret; +} #else - E_get_time(); +{ + return E_NO_SYS; +} #endif - if (Exit_events) goto end_handler; - break; - } - } -#ifdef TESTTIME - start = E_get_time(); - tmp_late = E_sub_time(start, stop); - Alarm(DEBUG, "Events: Low priority took %d %d to handle\n", tmp_late.sec, tmp_late.usec); -#endif + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +static int E_events_select_query(E_events * events, + E_priority query_lvl, + const sp_time * delta_timeout, + E_priority * queried_lvl) +{ + int ret = 0; + E_select_info * info = (E_select_info*) events->backend_info; + E_select_lvl_info * lvl_info = (E_select_lvl_info*) events->priorities[query_lvl].backend_lvl_info; + int nfds = 0; + fd_set * fd_sets[3] = { NULL, NULL, NULL }; + struct timeval timeout_dmy = { 0, 0 }; + struct timeval * timeout = &timeout_dmy; + E_watch_type fd_type; + E_fd_watch * fd_watch; + int tmp_fd; + stdit sit; + stdit tmp_it; + + /* set up select's inputs: nfds, fd_sets, timeout */ + + for (fd_type = 0; fd_type != NUM_FDTYPES; ++fd_type) { + + if (!stdskl_empty(&lvl_info->fd_skls[fd_type])) { /* make a tmp copy of lvl_info's fd set in info */ + + fd_sets[fd_type] = &info->tmp_fd_sets[fd_type]; + info->tmp_fd_sets[fd_type] = lvl_info->fd_sets[fd_type]; + tmp_fd = 1 + *(int*) stdskl_it_key(stdskl_last(&lvl_info->fd_skls[fd_type], &sit)); + nfds = STDMAX(nfds, tmp_fd); + assert(tmp_fd > 0); } - end_handler: - /* Clean up data structures for exit OR restart of handler loop */ - /* Actually nothing needs to be cleaned up to allow E_handle_events() - * to be called again. The events are still registered (or not registered) - * and the only state for the actual events loop is Exit_events which is reset - * in the for loop. - */ + } - return; + if (delta_timeout != NULL) { + timeout->tv_sec = delta_timeout->sec; + timeout->tv_usec = delta_timeout->usec; + + } else if (nfds != 0) { + timeout = NULL; + + } else { + ret = E_DEADEND; /* error on infinite timeout and no fds to watch */ + goto FAIL; + } + + /* query select */ + + if ((ret = select(nfds, fd_sets[READ_FD], fd_sets[WRITE_FD], fd_sets[EXCEPT_FD], timeout)) < 0) { + + ret = errno; + assert(ret != 0); + + switch (ret) { + + case EINTR: + ret = E_INTERRUPTED; /* will cause this fcn to be called again */ + goto FAIL; + + case EBADF: + ret = E_INVALID_PARAM; /* bad fd or fd w/ an error in a set */ + goto FAIL; + + case EINVAL: + assert(0); + ret = E_BUG; + goto FAIL; + + default: + ret = E_SYS_FAILURE; + goto FAIL; + } + } + + /* process the result sets */ + + for (fd_type = 0; ret != 0 && fd_type != NUM_FDTYPES; ++fd_type) { + + for (stdskl_begin(&lvl_info->fd_skls[fd_type], &sit); ret != 0 && !stdskl_is_end(&lvl_info->fd_skls[fd_type], &sit); stdskl_it_next(&sit)) { + + tmp_fd = *(int*) stdskl_it_key(&sit); + + if (!FD_ISSET(tmp_fd, fd_sets[fd_type])) { + continue; + } + + --ret; + + fd_watch = E_events_get_fd_watch(events, tmp_fd, fd_type, &tmp_it); + assert(fd_watch != NULL); + + if (fd_watch->w.noticed) { + continue; + } + + if (stddll_insert(&events->priorities[fd_watch->w.priority].notices, + stddll_end(&events->priorities[fd_watch->w.priority].notices, &fd_watch->w.notice_it), + &fd_watch) != 0) { + ret = E_ALLOC_FAILURE; + goto FAIL; + } + + fd_watch->w.noticed = STDTRUE; + } + } + + assert(ret == 0); + goto END; + + /* error handling and return */ + + FAIL: + assert(ret != 0); + + END: + return ret; } -void E_exit_events(void) +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +sp_time E_get_time(void) { - Alarm( EVENTS, "E_exit_events:\n"); - Exit_events = 1; + return E_events_get_time(&Events); } + +/*************************************************************************************************************************** + * NOTE: Bad implementation kept for backwards compatability -- should allow for negative times + ***************************************************************************************************************************/ + +sp_time E_sub_time(sp_time t, + sp_time delta_t) +{ + t.sec -= delta_t.sec; + t.usec -= delta_t.usec; + + if (t.usec < 0) { + t.usec += 1000000; + --t.sec; + } + + return t; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +sp_time E_add_time(sp_time t, + sp_time delta_t) +{ + t.sec += delta_t.sec; + t.usec += delta_t.usec; + + if (t.usec > 1000000) { + t.usec -= 1000000; + ++t.sec; + } + + return t; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_compare_time(sp_time t1, + sp_time t2) +{ + int ret = 0; + + if (t1.sec > t2.sec) ret = 1; + else if (t1.sec < t2.sec) ret = -1; + else if (t1.usec > t2.usec) ret = 1; + else if (t1.usec < t2.usec) ret = -1; + + return ret; +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +void E_delay(sp_time t) +#ifdef ARCH_PC_WIN95 +{ + SleepEx(t.sec * 1000 + t.usec / 1000, 0); +} +#else +{ + struct timeval tmp_t; + + tmp_t.tv_sec = t.sec; + tmp_t.tv_usec = t.usec; + + select(0, NULL, NULL, NULL, &tmp_t); +} +#endif + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_init(void) +{ + return E_events_init(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +void E_fini(void) +{ + E_events_fini(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +sp_time E_get_loop_time(void) +{ + return E_events_get_loop_time(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +sp_time E_update_loop_time(void) +{ + return E_events_update_loop_time(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_queue(E_time_callback_fcn fcn, int code, void *data, sp_time delta_time) +{ + return E_events_queue(&Events, fcn, code, data, delta_time); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_queue_absolute(E_time_callback_fcn fcn, int code, void *data, sp_time absolute_time) +{ + return E_events_queue_absolute(&Events, fcn, code, data, absolute_time); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_queue_periodic(E_time_callback_fcn fcn, int code, void *data, sp_time periodic_time) +{ + return E_events_queue_periodic(&Events, fcn, code, data, periodic_time); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_dequeue(E_time_callback_fcn fcn, int code, void *data) +{ + return E_events_dequeue(&Events, fcn, code, data); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_attach_fd(int fd, E_watch_type fd_type, E_fd_callback_fcn fcn, int code, void *data, E_priority priority) +{ + return E_events_attach_fd(&Events, fd, fd_type, fcn, code, data, priority); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_detach_fd(int fd, E_watch_type fd_type) +{ + return E_events_detach_fd(&Events, fd, fd_type); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_set_active_threshold(E_priority priority) +{ + return E_events_set_active_threshold(&Events, priority); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_num_active(E_priority priority) +{ + return E_events_num_active(&Events, priority); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_activate_fd(int fd, E_watch_type fd_type) +{ + return E_events_activate_fd(&Events, fd, fd_type); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_deactivate_fd(int fd, E_watch_type fd_type) +{ + return E_events_deactivate_fd(&Events, fd, fd_type); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_set_elevate_count(E_priority priority, unsigned count) +{ + return E_events_set_elevate_count(&Events, priority, count); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_get_elevate_count(E_priority priority, unsigned *count) +{ + return E_events_get_elevate_count(&Events, priority, count); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +int E_handle_events(void) +{ + return E_events_handle_events(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ + +void E_exit_events(void) +{ + E_events_exit_events(&Events); +} + +/*************************************************************************************************************************** + ***************************************************************************************************************************/ Modified: branches/events_testing/include/sp_events.h =================================================================== --- branches/events_testing/include/sp_events.h 2009-05-04 15:21:16 UTC (rev 411) +++ branches/events_testing/include/sp_events.h 2009-05-15 21:54:18 UTC (rev 412) @@ -32,74 +32,71 @@ * */ +#ifndef INC_SP_EVENTS +#define INC_SP_EVENTS +/* User level code should NOT include this file directly; include sp.h instead. + Spread daemon level code should include this file directly and NOT include sp.h +*/ -#ifndef INC_SP_EVENTS -#define INC_SP_EVENTS +/* Raise MAX_FD_EVENTS AND RECOMPILE events.c to handle more active + fd's. This number can limit the number of connections that can be + handled. +*/ -/* - User level code should NOT include this file directly. - Include sp.h instead. - Spread daemon level code should include this file directly and NOT include sp.h -*/ +#define MAX_FD_EVENTS 2000 +#define NUM_PRIORITY 3 +#define LOW_PRIORITY 0 +#define MEDIUM_PRIORITY 1 +#define HIGH_PRIORITY 2 -/* Raise this number AND RECOMPILE events.c to handle more active FD's. - * This number limits the number of connections that - * can be handled. - */ -#define MAX_FD_EVENTS 2000 +#define NUM_FDTYPES 3 +#define READ_FD 0 +#define WRITE_FD 1 +#define EXCEPT_FD 2 -#define NUM_PRIORITY 3 +typedef struct +{ + long sec; + long usec; -#define LOW_PRIORITY 0 -#define MEDIUM_PRIORITY 1 -#define HIGH_PRIORITY 2 +} sp_time; -#define NUM_FDTYPES 3 +/* Time routines */ -#define READ_FD 0 -#define WRITE_FD 1 -#define EXCEPT_FD 2 +sp_time E_get_time( void ); +sp_time E_sub_time( sp_time t, sp_time delta_t ); +sp_time E_add_time( sp_time t, sp_time delta_t ); +int E_compare_time( sp_time t1, sp_time t2 ); +void E_delay( sp_time t ); +/* Event routines */ -typedef struct dummy_time { - long sec; - long usec; -} sp_time; +int E_init( void ); -#ifndef NULL -#define NULL (void *)0 -#endif +int E_queue( void (* func)( int code, void *data ), int code, void *data, sp_time delta ); +int E_queue_absolute( void (* func)( int code, void *data ), int code, void *data, sp_time absolute ); +int E_queue_periodic( void (* func)( int code, void *data ), int code, void *data, sp_time period ); +int E_dequeue( void (* func)( int code, void *data ), int code, void *data ); -/* Event routines */ +int E_attach_fd( int fd, int fd_type, void (* func)( int fd, int code, void *data), + int code, void *data, int priority ); +int E_detach_fd( int fd, int fd_type ); -int E_init(void); -sp_time E_get_time(void); -sp_time E_sub_time( sp_time t, sp_time delta_t ); -sp_time E_add_time( sp_time t, sp_time delta_t ); -/* if t1 > t2 then returns 1; - if t1 < t2 then returns -1; - if t1 == t2 then returns 0; */ -int E_compare_time( sp_time t1, sp_time t2 ); -int E_queue( void (* func)( int code, void *data ), int code, void *data, - sp_time delta_time ); -/* Note: This does not dispose/free the data pointed at by the void - *data pointer */ -int E_dequeue( void (* func)( int code, void *data ), int code, - void *data ); -void E_delay( sp_time t ); +int E_set_active_threshold( int priority ); +int E_num_active( int priority ); -int E_attach_fd( int fd, int fd_type, - void (* func)( int fd, int code, void *data), int code, - void *data, int priority ); -int E_detach_fd( int fd, int fd_type ); -int E_set_active_threshold( int priority ); int E_activate_fd( int fd, int fd_type ); int E_deactivate_fd( int fd, int fd_type ); -int E_num_active( int priority ); -void E_handle_events(void); -void E_exit_events(void); +sp_time E_get_loop_time( void ); +sp_time E_update_loop_time( void ); -#endif /* INC_SP_EVENTS */ +int E_set_elevate_count( int priority, unsigned count ); +int E_get_elevate_count( int priority, unsigned *count ); + +int E_handle_events( void ); +void E_exit_events( void ); + +#endif