[Spread-cvs] commit: r412 - in branches/events_testing: daemon include
jschultz at spread.org
jschultz at spread.org
Fri May 15 17:54:18 EDT 2009
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 <info at spreadconcepts.com>
- *
- * All Rights Reserved.
- *
- * Major Contributor(s):
- * ---------------
- * Ryan Caudy rcaudy at gmail.com - contributions to process groups.
- * Claudiu Danilov claudiu at acm.org - scalable wide area support.
- * Cristina Nita-Rotaru crisn at cs.purdue.edu - group communication security.
- * Theo Schlossnagle jesus at omniti.com - Perl, autoconf, old skiplist.
- * Dan Schoenblum dansch at cnds.jhu.edu - 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
More information about the Spread-cvs
mailing list