text: /*
   * Copyright (c) 2001-2019 Julien Nadeau Carriere <vedge@csoft.net>
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions
   * are met:
   * 1. Redistributions of source code must retain the above copyright
   *    notice, this list of conditions and the following disclaimer.
   * 2. Redistributions in binary form must reproduce the above copyright
   *    notice, this list of conditions and the following disclaimer in the
   *    documentation and/or other materials provided with the distribution.
   * 
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
   * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE FOR
   * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
   * DAMAGES (INCLUDING BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
   * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
   * USE OF THIS SOFTWARE EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   */
  
  /*
   * Implementation of AG_Object events / virtual functions, as well as
   * the generic AG_EventLoop(3) interface.
   */
  
  #include <agar/core/core.h>
  
  #include <string.h>
  #include <stdarg.h>
  
  #include <agar/config/have_kqueue.h>
  #include <agar/config/have_timerfd.h>
  #include <agar/config/have_select.h>
  
  #if defined(HAVE_KQUEUE)
  # ifdef __NetBSD__
  #   define _NETBSD_SOURCE
  # endif
  # include <sys/types.h>
  # include <sys/event.h>
  # include <unistd.h>
  # include <errno.h>
  #endif
  #if defined(HAVE_TIMERFD)
  # include <sys/timerfd.h>
  # include <errno.h>
  #endif
  #if defined(HAVE_SELECT)
  # include <sys/types.h>
  # include <sys/time.h>
  # include <sys/select.h>
  # include <unistd.h>
  # include <errno.h>
  #endif
  
  /* Expensive debugging output related to event delivery. */
  /* #define DEBUG_EVENTS */
  
  AG_EventSource *_Nullable agEventSource = NULL;	/* Event source (thread-local) */
  #ifdef AG_THREADS
  AG_ThreadKey agEventSourceKey;
  #endif
  
  #ifdef HAVE_KQUEUE
  
  /* Size of kqueue input event buffer (in kevents). */
  # ifndef AG_KQ_EVBUFSIZE
  # define AG_KQ_EVBUFSIZE 2
  # endif
  
  /* Initial size of kqueue changes[] array. */
  # ifndef AG_KQ_INIT_MAXCHANGES
  # define AG_KQ_INIT_MAXCHANGES AG_MODEL
  # endif
  
  typedef struct ag_event_source_kqueue {
  	struct ag_event_source _inherit;  /* EventSource -> EventSourceKQUEUE */
  	struct kevent *_Nullable changes; /* Queued changes */
  	Uint                    nChanges;
  	Uint                  maxChanges;
  	struct kevent events[AG_KQ_EVBUFSIZE];  /* Input event buffer */
  	int fd;                                 /* kqueue() fd */
  	Uint32 _pad;
  } AG_EventSourceKQUEUE;
  
  static int GrowKqChangelist(AG_EventSourceKQUEUE *_Nonnull, Uint);
  #endif /* HAVE_KQUEUE */
  
  /* #define DEBUG_TIMERS */
  
  #ifdef __NetBSD__
  # define AG_EV_SET(kevp,a,b,c,d,e,f) EV_SET((kevp),(uintptr_t)(a),(b),(c),(d),(e),(intptr_t)(f))
  #else
  # define AG_EV_SET(kevp,a,b,c,d,e,f) EV_SET((kevp),(a),(b),(c),(d),(e),(f))
  #endif
  
  /* Import inlinables */
  #undef AG_INLINE_HEADER
  #include <agar/core/inline_event.h>
  
  #ifdef AG_DEBUG
  static __inline__ void
  InitDebugName(AG_Variable *_Nonnull V, const char *tag)
  {
  	Strlcpy(V->name, tag, sizeof(V->name));
  }
  #else
  # define InitDebugName(V,tag)
  #endif
  
  /* Initialize a pointer argument. */
  static __inline__ void
  InitPointerArg(AG_Variable *_Nonnull V, void *_Nullable p)
  {
  #ifdef AG_DEBUG
  	memset(V->name, '\0', sizeof(V->name));
  #else
  	V->name[0] = '\0';
  #endif
  	V->type = AG_VARIABLE_POINTER;
  #ifdef AG_THREADS
  	V->mutex = NULL;
  #endif
  	V->data.p = p;
  	V->info.pFlags = 0;
  #ifdef AG_DEBUG
  	V->vars.tqe_next = NULL;
  	V->vars.tqe_prev = NULL;
  #endif
  }
  
  static __inline__ void
  InitEvent(AG_Event *_Nonnull ev, AG_Object *_Nullable ob)
  {
  #ifdef AG_DEBUG
  	memset(ev->name, '\0', sizeof(ev->name));
  #else
  	ev->name[0] = '\0';
  #endif
  	ev->fn = NULL;
  	ev->argc = 1;
  	ev->argc0 = 1;
  	InitPointerArg(&ev->argv[0], ob);
  	InitDebugName (&ev->argv[0], "self");
  #ifdef AG_DEBUG
  	ev->events.tqe_next = NULL;
  	ev->events.tqe_prev = NULL;
  #endif
  }
  
  /* Initialize an AG_Event structure. */
  void
  AG_EventInit(AG_Event *_Nonnull ev)
  {
  	InitEvent(ev, NULL);
  }
  
  /* Initialize an AG_Event structure with the specified arguments. */
  void
  AG_EventArgs(AG_Event *ev, const char *fmt, ...)
  {
  	va_list ap;
  
  	InitEvent(ev, NULL);
  	if (fmt) {
  		va_start(ap, fmt);
  		AG_EventGetArgs(ev, fmt, ap);
  		va_end(ap);
  	}
  	ev->argc0 = ev->argc;
  }
  
  /* Return a newly allocated AG_Event. */
  AG_Event *
  AG_EventNew(AG_EventFn fn, void *obj, const char *fmt, ...)
  {
  	AG_Event *ev;
  	va_list ap;
  
  	ev = AG_Malloc(sizeof(AG_Event));
  	InitEvent(ev, obj);
  	ev->fn = fn;
  	if (fmt) {
  		va_start(ap, fmt);
  		AG_EventGetArgs(ev, fmt, ap);
  		va_end(ap);
  	}
  	return (ev);
  }
  
  #if AG_MODEL != AG_SMALL
  /* Return a newly-allocated duplicate of the given AG_Event. */
  AG_Event *
  AG_EventDup(const AG_Event *event)
  {
  	AG_Event *ev;
  
  	ev = Malloc(sizeof(AG_Event));
  	AG_EventCopy(ev, event);
  	return (ev);
  }
  #endif /* !AG_SMALL */
  
  /*
   * Copy the name, callback pointer, arguments and inheritable flags
   * from a source AG_Event to a destination AG_Event.
   */
  void
  AG_EventCopy(AG_Event *dst, const AG_Event *src)
  {
  	int i;
  
  	memcpy(dst->name, src->name, sizeof(dst->name));
  	dst->fn = src->fn;
  	dst->argc = src->argc;
  	dst->argc0 = src->argc0;
  	for (i = 0; i < src->argc; i++) {
  		AG_CopyVariable(&dst->argv[i], &src->argv[i]);
  	}
  	dst->events.tqe_next = NULL;
  	dst->events.tqe_prev = NULL;
  }
  
  /*
   * Configure an event handler routine for a given event.
   * If a handler routine already exists, replace it.
   */
  AG_Event *
  AG_SetEvent(void *p, const char *name, AG_EventFn fn, const char *fmt, ...)
  {
  	AG_Object *ob = p;
  	AG_Event *ev;
  
  	AG_ObjectLock(ob);
  
  	if (name != NULL) {
  		TAILQ_FOREACH(ev, &ob->events, events)
  			if (strcmp(ev->name, name) == 0)
  				break;
  	} else {
  		ev = NULL;
  	}
  	if (ev == NULL) {
  		ev = Malloc(sizeof(AG_Event));
  		InitEvent(ev, ob);
  		if (name != NULL) {
  			if (Strlcpy(ev->name, name, sizeof(ev->name)) >= sizeof(ev->name))
  				AG_FatalError("Event name too big");
  		} else {
  			ev->name[0] = '\0';
  		}
  		TAILQ_INSERT_TAIL(&ob->events, ev, events);
  	} else {
  		ev->argc = 1;
  		ev->argc0 = 1;
  	}
  	InitPointerArg(&ev->argv[0], ob);
  	InitDebugName (&ev->argv[0], "self");
  	ev->fn = fn;
  
  	if (fmt) {
  		va_list ap;
  
  		va_start(ap, fmt);
  		AG_EventGetArgs(ev, fmt, ap);
  		va_end(ap);
  	}
  	ev->argc0 = ev->argc;
  
  	AG_ObjectUnlock(ob);
  	return (ev);
  }
  
  /*
   * Configure an event handler routine for a given event.
   * If a handler routine already exists, don't replace it.
   */
  AG_Event *
  AG_AddEvent(void *p, const char *name, AG_EventFn fn, const char *fmt, ...)
  {
  	AG_Object *ob = p;
  	AG_Event *ev, *evOther;
  
  	AG_ObjectLock(ob);
  
  	ev = Malloc(sizeof(AG_Event));
  	InitEvent(ev, ob);
  
  	if (name != NULL) {
  		TAILQ_FOREACH(evOther, &ob->events, events) {
  			if (strcmp(evOther->name, name) == 0)
  				break;
  		}
  		if (Strlcpy(ev->name, name, sizeof(ev->name)) >= sizeof(ev->name))
  			AG_FatalError("Event name too big");
  	} else {
  		ev->name[0] = '\0';
  	}
  
  	ev->fn = fn;
  
  	if (fmt) {
  		va_list ap;
  
  		va_start(ap, fmt);
  		AG_EventGetArgs(ev, fmt, ap);
  		va_end(ap);
  	}
  	ev->argc0 = ev->argc;
  
  	TAILQ_INSERT_TAIL(&ob->events, ev, events);
  	AG_ObjectUnlock(ob);
  	return (ev);
  }
  
  /* Delete an event handler by name. */
  void
  AG_UnsetEvent(void *p, const char *name)
  {
  	AG_Object *ob = p;
  	AG_Event *ev;
  
  	AG_ObjectLock(ob);
  	TAILQ_FOREACH(ev, &ob->events, events) {
  		if (strcmp(name, ev->name) == 0)
  			break;
  	}
  	if (ev == NULL) {
  		goto out;
  	}
  	TAILQ_REMOVE(&ob->events, ev, events);
  	free(ev);
  out:
  	AG_ObjectUnlock(ob);
  }
  
  /* Delete an event handler by reference. */
  void
  AG_UnsetEventByPtr(void *p, AG_Event *ev)
  {
  	AG_Object *ob = p;
  
  	AG_ObjectLock(ob);
  	TAILQ_REMOVE(&ob->events, ev, events);
  	AG_ObjectUnlock(ob);
  
  	free(ev);
  }
  
  /* Look up an AG_Event by name. */
  AG_Event *
  AG_FindEventHandler(void *p, const char *name)
  {
  	AG_Object *ob = p;
  	AG_Event *ev;
  	
  	AG_ObjectLock(ob);
  	TAILQ_FOREACH(ev, &ob->events, events) {
  		if (strcmp(name, ev->name) == 0)
  			break;
  	}
  	AG_ObjectUnlock(ob);
  	return (ev);
  }
  
  #ifdef AG_TIMERS
  /*
   * Timeout callback for scheduled events.
   */
  static Uint32
  EventTimeout(AG_Timer *_Nonnull to, AG_Event *_Nonnull event)
  {
  	AG_Object *obj = AG_OBJECT_SELF();
  	const char *eventName = AG_STRING(1);
  	AG_Event *ev;
  
  # ifdef DEBUG_EVENTS
  	Debug(obj, "Event <%s> timeout (%u ticks)\n", eventName,
  	    (Uint)to->ival);
  # endif
  	TAILQ_FOREACH(ev, &obj->events, events) {
  		if (strcmp(eventName, ev->name) == 0)
  			break;
  	}
  	if (ev == NULL) {
  		return (0);
  	}
  	/* Invoke the event handler routine. */
  	if (ev->fn != NULL) {
  		ev->fn(ev);
  	}
  	return (0);
  }
  #endif /* AG_TIMERS */
  
  /*
   * Variant of AG_PostEvent() which accepts a resolved AG_Event pointer
   * (as opposed to an event name string).
   */
  void
  AG_PostEventByPtr(void *pObj, AG_Event *ev, const char *fmt, ...)
  {
  	AG_Object *obj = pObj;
  	va_list ap;
  
  # ifdef DEBUG_EVENTS
  	Debug(obj, "Event %p posted\n", ev);
  # endif
  	AG_ObjectLock(obj);
  # if AG_MODEL == AG_SMALL
  	{
  		AG_Event *evTmp = Malloc(sizeof(AG_Event));
  
  		memcpy(evTmp, ev, sizeof(AG_Event));
  		if (fmt) {
  			va_start(ap, fmt);
  			AG_EventGetArgs(evTmp, fmt, ap);
  			va_end(ap);
  		}
  		if (evTmp->fn != NULL) {
  			evTmp->fn(evTmp);
  		}
  		free(evTmp);
  	}
  # else /* MEDIUM or LARGE */
  	{
  		AG_Event evTmp;				/* Fits the stack */
  
  		memcpy(&evTmp, ev, sizeof(AG_Event));
  		if (fmt) {
  			va_start(ap, fmt);
  			AG_EventGetArgs(&evTmp, fmt, ap);
  			va_end(ap);
  		}
  		if (evTmp.fn != NULL)
  			evTmp.fn(&evTmp);
  	}
  # endif /* MEDIUM or LARGE */
  
  	AG_ObjectUnlock(obj);
  }
  
  /*
   * Post an event (by name) to an object. If fmt is given, append the
   * given arguments (specified in the same format as AG_SetEvent(3)),
   * to the end of the argument vector.
   */
  void
  AG_PostEvent(void *pObj, const char *evname, const char *fmt, ...)
  {
  	AG_Object *obj = pObj;
  	AG_Event *ev;
  	va_list ap;
  
  #ifdef AG_DEBUG
  	if (obj == NULL) { AG_FatalError("NULL object"); }
  #endif
  #ifdef DEBUG_EVENTS
  	Debug(obj, "PostEvent <%s>\n", evname);
  #endif
  	AG_ObjectLock(obj);
  	TAILQ_FOREACH(ev, &obj->events, events) {
  		if (strcmp(evname, ev->name) != 0)
  			continue;
  #if AG_MODEL == AG_SMALL
  		{
  			AG_Event *evTmp = Malloc(sizeof(AG_Event));
  
  			memcpy(evTmp, ev, sizeof(AG_Event));
  			if (fmt) {
  				va_start(ap, fmt);
  				AG_EventGetArgs(evTmp, fmt, ap);
  				va_end(ap);
  			}
  			if (evTmp->fn != NULL) {
  				evTmp->fn(evTmp);
  			}
  			free(evTmp);
  		}
  #else /* MEDIUM or LARGE */
  		{
  			AG_Event evTmp;			/* Fits the stack */
  
  			memcpy(&evTmp, ev, sizeof(AG_Event));
  			if (fmt) {
  				va_start(ap, fmt);
  				AG_EventGetArgs(&evTmp, fmt, ap);
  				va_end(ap);
  			}
  			if (evTmp.fn != NULL)
  				evTmp.fn(&evTmp);
  		}
  #endif /* MEDIUM or LARGE */
  	}
  	AG_ObjectUnlock(obj);
  }
  
  /*
   * Parse a format string of AG_Event-style arguments into the argument
   * vector of ev. The parser macros are defined in <core/event.h>.
   */
  void
  AG_EventGetArgs(AG_Event *ev, const char *fmt, va_list ap)
  {
  	if (fmt) {
  		const char *c = (const char *)fmt;
  
  		while (*c != '\0')
  			AG_EVENT_PUSH_ARG(ap, ev);
  	}
  }
  
  /*
   * Forward an event to an object. The original arguments are copied
   * as is, except for Pointer 0 (SELF) which becomes pObj.
   */
  void
  AG_ForwardEvent(void *pObj, const AG_Event *event)
  {
  	AG_Object *obj = pObj;
  	AG_Event *ev;
  
  #ifdef DEBUG_EVENTS
  	Debug(obj, "Event <%s> forwarded\n", event->name);
  #endif
  	AG_ObjectLock(obj);
  	TAILQ_FOREACH(ev, &obj->events, events) {
  		if (strcmp(event->name, ev->name) != 0)
  			continue;
  #if AG_MODEL == AG_SMALL
  		{
  			AG_Event *evTmp = Malloc(sizeof(AG_Event));
  
  			memcpy(evTmp, event, sizeof(AG_Event));
  			InitPointerArg(&evTmp->argv[0], obj);
  			InitDebugName (&evTmp->argv[0], "self");
  			if (ev->fn != NULL) {
  				ev->fn(evTmp);
  			}
  			free(evTmp);
  		}
  #else /* MEDIUM or LARGE */
  		{
  			AG_Event evTmp;			/* Fits the stack */
  
  			memcpy(&evTmp, event, sizeof(AG_Event));
  			InitPointerArg(&evTmp.argv[0], obj);
  			InitDebugName (&evTmp.argv[0], "self");
  
  			if (ev->fn != NULL)
  				ev->fn(&evTmp);
  		}
  #endif /* MEDIUM or LARGE */
  	}
  	AG_ObjectUnlock(obj);
  }
  
  #ifdef AG_TIMERS
  /*
   * Schedule the execution of the named event in the given number
   * of AG_Time(3) ticks.
   *
   * The argument vector passed to the event handler function contains
   * the AG_SetEvent() arguments, and any arguments specified here are
   * appended to that list.
   */
  int
  AG_SchedEvent(void *pObj, Uint32 ticks, const char *evname, const char *fmt, ...)
  {
  	AG_Object *obj = pObj;
  	AG_Event *ev;
  	AG_Timer *to;
  	va_list ap;
  
  	if ((to = TryMalloc(sizeof(AG_Timer))) == NULL) {
  		return (-1);
  	}
  	AG_InitTimer(to, evname, AG_TIMER_AUTO_FREE);
  
  	AG_LockTiming();
  	AG_ObjectLock(obj);
  	
  	if (AG_AddTimer(obj, to, ticks, EventTimeout, "%s", evname) == -1) {
  		free(to);
  		goto fail;
  	}
  	ev = &to->fnEvent;
  	AG_EventInit(ev);
  	ev->argv[0].data.p = obj;
  	if (fmt) {
  		va_start(ap, fmt);
  		AG_EventGetArgs(ev, fmt, ap);
  		va_end(ap);
  	}
  	ev->argc0 = ev->argc;
  
  	AG_UnlockTiming();
  	AG_ObjectUnlock(obj);
  	return (0);
  fail:
  	AG_UnlockTiming();
  	AG_ObjectUnlock(obj);
  	return (-1);
  }
  #endif /* AG_TIMERS */
  
  #ifdef AG_EVENT_LOOP
  /*
   * Create a new event source.
   */
  static AG_EventSource *_Nullable
  CreateEventSource(void)
  {
  # ifdef HAVE_KQUEUE
  	AG_EventSourceKQUEUE *kq = TryMalloc(sizeof(AG_EventSourceKQUEUE));
  	AG_EventSource *src = (AG_EventSource *)kq;
  # else
  	AG_EventSource *src = TryMalloc(sizeof(AG_EventSource));
  # endif
  	if (src == NULL) {
  		return (NULL);
  	}
  	src->flags = 0;
  	src->breakReq = 0;
  # ifdef AG_TIMERS
  	src->addTimerFn = NULL;
  	src->delTimerFn = NULL;
  # endif
  	TAILQ_INIT(&src->prologues);
  	TAILQ_INIT(&src->epilogues);
  	TAILQ_INIT(&src->spinners);
  	TAILQ_INIT(&src->sinks);
  	src->returnCode = 0;
  	memset(src->caps, 0, sizeof(src->caps));
  
  # if defined(HAVE_KQUEUE)
  	if ((kq->fd = kqueue()) == -1) {
  		AG_SetError("kqueue: %s", AG_Strerror(errno));
  		return (NULL);
  	}
  	kq->changes = NULL;
  	kq->nChanges = 0;
  	kq->maxChanges = 0;
  	memset(kq->events, 0, AG_KQ_EVBUFSIZE*sizeof(struct kevent));
  	src->sinkFn = AG_EventSinkKQUEUE;
  #  ifdef AG_TIMERS
  	src->addTimerFn = AG_AddTimerKQUEUE;
  	src->delTimerFn = AG_DelTimerKQUEUE;
  #  endif
  	src->caps[AG_SINK_TIMER] = 1;		/* Provides timers internally */
  	src->caps[AG_SINK_READ] = 1;
  	src->caps[AG_SINK_WRITE] = 1;
  	src->caps[AG_SINK_FSEVENT] = 1;
  	src->caps[AG_SINK_PROCEVENT] = 1;
  	if (GrowKqChangelist(kq, AG_KQ_INIT_MAXCHANGES) == -1) {
  		AG_FatalError("GrowKqChangelist");
  	}
  # elif defined(HAVE_TIMERFD)
  	src->sinkFn = AG_EventSinkTIMERFD;
  #  ifdef AG_TIMERS
  	src->addTimerFn = AG_AddTimerTIMERFD;
  	src->delTimerFn = AG_DelTimerTIMERFD;
  #  endif
  	src->caps[AG_SINK_TIMER] = 1;		/* Provides timers internally */
  	src->caps[AG_SINK_READ] = 1;
  	src->caps[AG_SINK_WRITE] = 1;
  # elif defined(HAVE_SELECT) && !defined(AG_THREADS)
  	src->sinkFn = AG_EventSinkTIMEDSELECT;
  	src->caps[AG_SINK_READ] = 1;
  	src->caps[AG_SINK_WRITE] = 1;
  # elif defined(HAVE_SELECT) && defined(AG_THREADS)
  	src->sinkFn = AG_EventSinkSELECT;
  	src->caps[AG_SINK_READ] = 1;
  	src->caps[AG_SINK_WRITE] = 1;
  # else
  	src->sinkFn = AG_EventSinkSPINNER;
  # endif
  # ifdef AG_TIMERS
  	if (agSoftTimers) {			/* Force soft timers */
  		src->addTimerFn = NULL;
  		src->delTimerFn = NULL;
  		src->caps[AG_SINK_TIMER] = 0;
  	}
  # else
  	src->caps[AG_SINK_TIMER] = 0;
  # endif
  	return (src);
  }
  
  /* Free all resources allocated by an event source. */
  static void
  DestroyEventSource(void *_Nullable pEventSource)
  {
  	AG_EventSource *src = pEventSource;
  	AG_EventSink *es, *esNext;
  	
  	if (agEventSource == NULL)
  		return;
  
  # ifdef HAVE_KQUEUE
  	{
  		AG_EventSourceKQUEUE *kq = pEventSource;
  
  		if (kq->fd != -1) {
  			close(kq->fd);
  		}
  		Free(kq->changes);
  	}
  # endif
  	for (es = TAILQ_FIRST(&src->prologues);
  	     es != TAILQ_END(&src->prologues);
  	     es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		free(es);
  	}
  	for (es = TAILQ_FIRST(&src->epilogues);
  	     es != TAILQ_END(&src->epilogues);
  	     es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		free(es);
  	}
  	for (es = TAILQ_FIRST(&src->spinners);
  	     es != TAILQ_END(&src->spinners);
  	     es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		free(es);
  	}
  	for (es = TAILQ_FIRST(&src->sinks);
  	     es != TAILQ_END(&src->sinks);
  	     es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		free(es);
  	}
  	free(src);
  }
  
  int
  AG_InitEventSubsystem(Uint flags)
  {
  	/* Initialize the main thread's event source. */
  	agEventSource = NULL;
  #ifdef AG_THREADS
  	if (AG_ThreadKeyTryCreate(&agEventSourceKey, DestroyEventSource) == -1)
  		return (-1);
  #endif
  	if ((agEventSource = AG_GetEventSource()) == NULL) {
  		return (-1);
  	}
  	return (0);
  }
  
  void
  AG_DestroyEventSubsystem(void)
  {
  	if (agEventSource != NULL) {
  		DestroyEventSource(agEventSource);
  		agEventSource = NULL;
  	}
  }
  
  /* Return the calling thread's effective event source. */
  AG_EventSource *
  AG_GetEventSource(void)
  {
  	AG_EventSource *src;
  
  # ifdef AG_THREADS
  	if ((src = AG_ThreadKeyGet(agEventSourceKey)) != NULL && src != NULL)
  		return (src);
  # else
  	if (agEventSource != NULL)
  		return (agEventSource);
  # endif
  	if ((src = CreateEventSource()) == NULL)
  		AG_FatalError(NULL);
  # ifdef AG_THREADS
  	AG_ThreadKeySet(agEventSourceKey, src);
  # else
  	agEventSource = src;
  # endif
  	return (src);
  }
  
  # ifdef HAVE_KQUEUE
  /*
   * Routines for translating between AG_EventSink and kqueue types.
   */
  static __inline__ enum ag_event_sink_type _Const_Attribute
  GetSinkType(int filter)
  {
  	switch (filter) {
  	case EVFILT_TIMER:	return (AG_SINK_TIMER);
  	case EVFILT_READ:	return (AG_SINK_READ);
  	case EVFILT_WRITE:	return (AG_SINK_WRITE);
  	case EVFILT_VNODE:	return (AG_SINK_FSEVENT);
  	case EVFILT_PROC:	return (AG_SINK_PROCEVENT);
  	default:		return (AG_SINK_NONE);
  	}
  }
  static Uint _Const_Attribute
  GetKqFilterFlags(Uint flags)
  {
  	Uint fflags = 0;
  	if (flags & AG_FSEVENT_DELETE) { fflags |= NOTE_DELETE; }
  	if (flags & AG_FSEVENT_WRITE)  { fflags |= NOTE_WRITE;  }
  	if (flags & AG_FSEVENT_EXTEND) { fflags |= NOTE_EXTEND; }
  	if (flags & AG_FSEVENT_ATTRIB) { fflags |= NOTE_ATTRIB; }
  	if (flags & AG_FSEVENT_LINK)   { fflags |= NOTE_LINK;   }
  	if (flags & AG_FSEVENT_RENAME) { fflags |= NOTE_RENAME; }
  	if (flags & AG_FSEVENT_REVOKE) { fflags |= NOTE_REVOKE; }
  	if (flags & AG_PROCEVENT_EXIT) { fflags |= NOTE_EXIT; }
  	if (flags & AG_PROCEVENT_FORK) { fflags |= NOTE_FORK; }
  	if (flags & AG_PROCEVENT_EXEC) { fflags |= NOTE_EXEC; }
  	return (fflags);
  }
  static Uint _Const_Attribute
  GetSinkFlags(Uint fflags)
  {
  	Uint flags = 0;
  	if (fflags & NOTE_DELETE) { fflags |= AG_FSEVENT_DELETE; }
  	if (fflags & NOTE_WRITE)  { fflags |= AG_FSEVENT_WRITE;  }
  	if (fflags & NOTE_EXTEND) { fflags |= AG_FSEVENT_EXTEND; }
  	if (fflags & NOTE_ATTRIB) { fflags |= AG_FSEVENT_ATTRIB; }
  	if (fflags & NOTE_LINK)   { fflags |= AG_FSEVENT_LINK;   }
  	if (fflags & NOTE_RENAME) { fflags |= AG_FSEVENT_RENAME; }
  	if (fflags & NOTE_REVOKE) { fflags |= AG_FSEVENT_REVOKE; }
  	if (fflags & NOTE_EXIT) { fflags |= AG_PROCEVENT_EXIT; }
  	if (fflags & NOTE_FORK) { fflags |= AG_PROCEVENT_FORK; }
  	if (fflags & NOTE_EXEC) { fflags |= AG_PROCEVENT_EXEC; }
  	return (flags);
  }
  # endif /* HAVE_KQUEUE */
  
  /*
   * Add/remove an event processing prologue. The function will be invoked
   * only once at the beginning of AG_EventLoop().
   */
  AG_EventSink *
  AG_AddEventPrologue(AG_EventSinkFn fn, const char *fnArgs, ...)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es;
  	va_list ap;
  
  	if ((es = TryMalloc(sizeof(AG_EventSink))) == NULL) {
  		return (NULL);
  	}
  	es->type = AG_SINK_PROLOGUE;
  	es->fn = fn;
  	InitEvent(&es->fnArgs, NULL);
  	if (fnArgs) {
  		va_start(ap, fnArgs);
  		AG_EventGetArgs(&es->fnArgs, fnArgs, ap);
  		va_end(ap);
  	}
  	es->fnArgs.argc0 = es->fnArgs.argc;
  	TAILQ_INSERT_TAIL(&src->prologues, es, sinks);
  	return (es);
  }
  void
  AG_DelEventPrologue(AG_EventSink *es)
  {
  	AG_EventSource *src = AG_GetEventSource();
  
  # ifdef AG_DEBUG
  	if (es->type != AG_SINK_PROLOGUE)
  		AG_FatalError("AG_DelEventPrologue");
  # endif
  	TAILQ_REMOVE(&src->prologues, es, sinks);
  	free(es);
  }
  
  /*
   * Add/remove an event sink epilogue. The function will be invoked
   * after all incoming / queued events have been processed.
   */
  AG_EventSink *
  AG_AddEventEpilogue(AG_EventSinkFn fn, const char *fnArgs, ...)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es;
  	va_list ap;
  
  	if ((es = TryMalloc(sizeof(AG_EventSink))) == NULL) {
  		return (NULL);
  	}
  	es->type = AG_SINK_EPILOGUE;
  	es->fn = fn;
  	InitEvent(&es->fnArgs, NULL);
  	if (fnArgs) {
  		va_start(ap, fnArgs);
  		AG_EventGetArgs(&es->fnArgs, fnArgs, ap);
  		va_end(ap);
  	}
  	es->fnArgs.argc0 = es->fnArgs.argc;
  	TAILQ_INSERT_TAIL(&src->epilogues, es, sinks);
  	return (es);
  }
  void
  AG_DelEventEpilogue(AG_EventSink *es)
  {
  	AG_EventSource *src = AG_GetEventSource();
  
  # ifdef AG_DEBUG
  	if (es->type != AG_SINK_EPILOGUE)
  		AG_FatalError("AG_DelEventEpilogue");
  # endif
  	TAILQ_REMOVE(&src->epilogues, es, sinks);
  	free(es);
  }
  
  /*
   * Add/remove a spinning event sink. If at least one spinning sink exists,
   * AG_EventLoop() will invoke it repeatedly and force non-blocking operation
   * of subsequent polling methods.
   */
  AG_EventSink *
  AG_AddEventSpinner(AG_EventSinkFn fn, const char *fnArgs, ...)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es;
  	va_list ap;
  
  	if ((es = TryMalloc(sizeof(AG_EventSink))) == NULL) {
  		return (NULL);
  	}
  	es->type = AG_SINK_SPINNER;
  	es->fn = fn;
  	InitEvent(&es->fnArgs, NULL);
  	if (fnArgs) {
  		va_start(ap, fnArgs);
  		AG_EventGetArgs(&es->fnArgs, fnArgs, ap);
  		va_end(ap);
  	}
  	es->fnArgs.argc0 = es->fnArgs.argc;
  	TAILQ_INSERT_TAIL(&src->spinners, es, sinks);
  	return (es);
  }
  void
  AG_DelEventSpinner(AG_EventSink *es)
  {
  	AG_EventSource *src = AG_GetEventSource();
  
  # ifdef AG_DEBUG
  	if (es->type != AG_SINK_SPINNER)
  		AG_FatalError("AG_DelEventSpinner");
  # endif
  	TAILQ_REMOVE(&src->spinners, es, sinks);
  	free(es);
  }
  
  /*
   * Add/remove a low-level event sink. The function will be called
   * whenever the specified event occurs.
   */
  AG_EventSink *
  AG_AddEventSink(enum ag_event_sink_type type, int ident, Uint flags,
      AG_EventSinkFn fn, const char *fnArgs, ...)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es;
  # ifdef HAVE_KQUEUE
  	AG_EventSourceKQUEUE *kq = (AG_EventSourceKQUEUE *)src;
  	struct kevent *kev;
  # endif
  	va_list ap;
  
  	if (type >= AG_SINK_LAST || !src->caps[type]) {
  		AG_SetErrorV("E1", "No such event type");
  		return (NULL);
  	}
  	if ((es = TryMalloc(sizeof(AG_EventSink))) == NULL) {
  		return (NULL);
  	}
  	es->type = type;
  	es->ident = ident;
  	es->flags = flags;
  
  # ifdef HAVE_KQUEUE
  	if (kq->nChanges+1 > kq->maxChanges &&
  	    GrowKqChangelist(kq, kq->nChanges+1) == -1) {
  		free(es);
  		return (NULL);
  	}
  	kev = &kq->changes[kq->nChanges++];
  	switch (type) {
  	case AG_SINK_READ:
  		AG_EV_SET(kev, ident, EVFILT_READ, EV_ADD|EV_ENABLE, 0, 0, es);
  		break;
  	case AG_SINK_WRITE:
  		AG_EV_SET(kev, ident, EVFILT_WRITE, EV_ADD|EV_ENABLE, 0, 0, es);
  		break;
  	case AG_SINK_FSEVENT:
  		AG_EV_SET(kev, ident, EVFILT_VNODE, EV_ADD|EV_ENABLE,
  		    GetKqFilterFlags(flags), 0, es);
  		break;
  	case AG_SINK_PROCEVENT:
  		AG_EV_SET(kev, ident, EVFILT_PROC, EV_ADD|EV_ENABLE,
  		    GetKqFilterFlags(flags), 0, es);
  		break;
  	default:
  		kq->nChanges--;
  		break;
  	}
  # endif /* HAVE_KQUEUE */
  
  	es->fn = fn;
  	InitEvent(&es->fnArgs, NULL);
  	if (fnArgs) {
  		va_start(ap, fnArgs);
  		AG_EventGetArgs(&es->fnArgs, fnArgs, ap);
  		va_end(ap);
  	}
  	es->fnArgs.argc0 = es->fnArgs.argc;
  	TAILQ_INSERT_TAIL(&src->sinks, es, sinks);
  	return (es);
  }
  
  # ifdef HAVE_KQUEUE
  static int
  GrowKqChangelist(AG_EventSourceKQUEUE *_Nonnull kq, Uint n)
  {
  	struct kevent *changesNew;
  
  	if ((changesNew = TryRealloc(kq->changes, n*sizeof(struct kevent)))
  	    == NULL) {
  		return (-1);
  	}
  	kq->changes = changesNew;
  	kq->maxChanges = n;
  	return (0);
  }
  # endif /* HAVE_KQUEUE */
  
  void
  AG_DelEventSink(AG_EventSink *es)
  {
  	AG_EventSource *src = AG_GetEventSource();
  # ifdef HAVE_KQUEUE
  	AG_EventSourceKQUEUE *kq = (AG_EventSourceKQUEUE *)src;
  	struct kevent *kev;
  
  	if (kq->nChanges+1 > kq->maxChanges &&
  	    GrowKqChangelist(kq, kq->nChanges+1) == -1) {
  		AG_FatalError(NULL);
  	}
  	kev = &kq->changes[kq->nChanges++];
  
  	switch (es->type) {
  	case AG_SINK_READ:
  		AG_EV_SET(kev, es->ident, EVFILT_READ, EV_DELETE, 0, 0, NULL);
  		break;
  	case AG_SINK_WRITE:
  		AG_EV_SET(kev, es->ident, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
  		break;
  	case AG_SINK_FSEVENT:
  		AG_EV_SET(kev, es->ident, EVFILT_VNODE, EV_DELETE,
  		    GetKqFilterFlags(es->flags), 0, NULL);
  		break;
  	case AG_SINK_PROCEVENT:
  		AG_EV_SET(kev, es->ident, EVFILT_PROC, EV_DELETE,
  		    GetKqFilterFlags(es->flags), 0, NULL);
  		break;
  	default:
  		kq->nChanges--;
  		break;
  	}
  # endif /* HAVE_KQUEUE */
  
  	TAILQ_REMOVE(&src->sinks, es, sinks);
  	free(es);
  }
  
  void
  AG_DelEventSinksByIdent(enum ag_event_sink_type type, int ident, Uint flags)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es, *esNext;
  
  	for (es = TAILQ_FIRST(&src->sinks); es != TAILQ_END(&src->sinks); es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		if (es->type == type &&
  		    es->ident == ident &&
  		    es->flags == flags)
  			AG_DelEventSink(es);
  	}
  }
  
  # ifdef HAVE_KQUEUE
  /*
   * Standard event sink using kqueue(2), commonly found on modern BSD
   * derived operating systems. 
   */
  int
  AG_EventSinkKQUEUE(void)
  {
  	AG_EventSourceKQUEUE *kq = (AG_EventSourceKQUEUE *)agEventSource;
  	int rv, i;
  	struct timespec timeo, *pTimeo;
  
  restart:
  	if (!TAILQ_EMPTY(&agEventSource->spinners)) {
  		timeo.tv_sec = 0;
  		timeo.tv_nsec = 0L;
  		pTimeo = &timeo;
  	} else {
  		pTimeo = NULL;
  	}
  #  ifdef DEBUG_TIMERS
  	for (i = 0; i < kq->nChanges; i++) {
  		struct kevent *chg = &kq->changes[i];
  		Verbose("changes[%d]: f=%d i=%u f=0x%x ff=0x%x u=%p\n",
  		    i, (int)chg->filter,
  		    (Uint)chg->ident, chg->flags, chg->fflags,
  		    chg->udata);
  	}
  #  endif
  	rv = kevent(kq->fd, kq->changes, kq->nChanges, kq->events,
  	            AG_KQ_EVBUFSIZE, pTimeo);
  	if (rv < 0) {
  		if (errno == EINTR) {
  			goto restart;
  		}
  		AG_SetError("kevent(): %s", AG_Strerror(errno));
  		return (-1);
  	}
  	kq->nChanges = 0;
  
  #  ifdef AG_TIMERS
  	/* 1. Process timer expirations. */
  	AG_LockTiming();
  	for (i = 0; i < rv; i++) {
  		struct kevent *kev = &kq->events[i];
  		const enum ag_event_sink_type esType = GetSinkType(kev->filter);
  		Uint32 rvt;
  		AG_Timer *to;
  		AG_Object *ob;
  
  		if (kev->flags & EV_ERROR) {
  #  ifdef DEBUG_TIMERS
  			Verbose("kevent (%ld,%d): %s\n", kev->ident, kev->filter,
  			    AG_Strerror((int)kev->data));
  #  endif
  			continue;
  		}
  		if (esType != AG_SINK_TIMER ||
  		    (to = (AG_Timer *)kev->udata) == NULL) {
  			continue;
  		}
  		rvt = to->fn(to, &to->fnEvent);
  		if (rvt > 0) {				/* Restart timer */
  			struct kevent *kev;
  #  ifdef DEBUG_TIMERS
  			Verbose("TIMER[%d] resetting t=+%u\n", to->id,
  			    (Uint)rvt);
  #  endif
  			if (kq->nChanges+1 > kq->maxChanges &&
  			    GrowKqChangelist(kq, kq->nChanges+1) == -1) {
  				AG_UnlockTiming();
  				return (-1);
  			}
  			kev = &kq->changes[kq->nChanges++];
  			AG_EV_SET(kev, to->id, EVFILT_TIMER,
  			    EV_ADD | EV_ENABLE | EV_ONESHOT, 0, (int)rvt, to);
  			to->ival = rvt;
  		} else {				/* Expire */
  #  ifdef DEBUG_TIMERS
  			Verbose("TIMER[%d] expired\n", to->id);
  #  endif
  			if ((ob = to->obj) == NULL) {
  				continue;
  			}
  			TAILQ_REMOVE(&ob->timers, to, pvt.timers);
  			if (TAILQ_EMPTY(&ob->timers)) {
  				TAILQ_REMOVE(&agTimerObjQ, ob, tobjs);
  			}
  			if (to->flags & AG_TIMER_AUTO_FREE) {
  				free(to);
  			} else {
  				to->ival = 0;
  				to->id = -1;
  				to->obj = NULL;
  			}
  			agTimerCount--;
  		}
  	}
  	AG_UnlockTiming();
  #  endif /* AG_TIMERS */
  
  	/* 2. Process I/O and other events. */
  	for (i = 0; i < rv; i++) {
  		struct kevent *kev = &kq->events[i];
  		const enum ag_event_sink_type esType = GetSinkType(kev->filter);
  		AG_EventSink *es;
  
  		switch (esType) {
  		case AG_SINK_READ:
  		case AG_SINK_WRITE:
  			es = (AG_EventSink *)kev->udata;
  			if (es) {
  				es->fn(es, &es->fnArgs);
  			}
  			break;
  		case AG_SINK_FSEVENT:
  		case AG_SINK_PROCEVENT:
  			es = (AG_EventSink *)kev->udata;
  			if (es) {
  				es->flagsMatched = GetSinkFlags(kev->fflags);
  				es->fn(es, &es->fnArgs);
  			}
  			break;
  		default:
  			break;
  		}
  	}
  	return (0);
  }
  
  #  ifdef AG_TIMERS
  /*
   * Add/remove a kqueue(2) based timer.
   */
  /*
   * XXX TODO only increment agTimerCount after timer has been actually
   * created (and make GenerateTimerID() pure).
   */
  static int
  GenerateTimerID(AG_Timer *_Nonnull to)
  {
  	AG_Object *obOther;
  	AG_Timer *toOther;
  	int id;
  
  gen_id:
  #   ifdef AG_DEBUG
  	if (agTimerCount+1 >= (AG_INT_MAX-1))
  		AG_FatalError("agTimerCount");
  #   endif
  	id = (int)++agTimerCount;			/* XXX */
  	TAILQ_FOREACH(obOther, &agTimerObjQ, tobjs) {
  		TAILQ_FOREACH(toOther, &obOther->timers, pvt.timers) {
  			if (toOther == to) { continue; }
  			if (toOther->id == id) {
  				id++;
  				goto gen_id;
  			}
  		}
  	}
  	return (id);
  }
  
  /* Create a new kernel-based timer with kqueue(2). */
  int
  AG_AddTimerKQUEUE(AG_Timer *to, Uint32 ival, int newTimer)
  {
  	AG_EventSourceKQUEUE *kq = (AG_EventSourceKQUEUE *)agEventSource;
  	
  	if (newTimer) {
  		to->id = GenerateTimerID(to);
  	}
  	if (newTimer || to->ival != ival) {
  		struct kevent *kev;
  #   ifdef DEBUG_TIMERS
  		Verbose("kevent: creating timer ID=%d ival=%d\n", to->id, (int)ival);
  #   endif
  		if (kq->nChanges+1 > kq->maxChanges &&
  		    GrowKqChangelist(kq, kq->nChanges+1) == -1) {
  			return (-1);
  		}
  		kev = &kq->changes[kq->nChanges++];
  		AG_EV_SET(kev, to->id, EVFILT_TIMER,
  		    EV_ADD | EV_ENABLE | EV_ONESHOT, 0, (int)ival, to);
  		to->ival = ival;
  	}
  	return (0);
  }
  
  /* Delete an active kqueue(2) timer. */
  void
  AG_DelTimerKQUEUE(AG_Timer *to)
  {
  	AG_EventSourceKQUEUE *kq = (AG_EventSourceKQUEUE *)agEventSource;
  	struct kevent *kev;
  
  	if (kq->nChanges+1 > kq->maxChanges &&
  	    GrowKqChangelist(kq, kq->nChanges+1) == -1) {
  		AG_FatalError(NULL);
  	}
  	kev = &kq->changes[kq->nChanges++];
  	AG_EV_SET(kev, to->id, EVFILT_TIMER, EV_DELETE, 0, 0, NULL);
  	agTimerCount--;
  }
  #  endif /* AG_TIMERS */
  # endif /* HAVE_KQUEUE */
  
  # ifdef HAVE_TIMERFD
  /*
   * Standard event sink using select(2) and fd-based timers,
   * usually available on Linux.
   */
  int
  AG_EventSinkTIMERFD(void)
  {
  	fd_set rdFds, wrFds;
  	int nFds, rv;
  	AG_EventSink *es, *esNext;
  	AG_Object *ob, *obNext;
  	AG_Timer *to, *toNext;
  	struct timeval timeo, *pTimeo;
  
  restart:
  	nFds = 0;
  	FD_ZERO(&rdFds);
  	FD_ZERO(&wrFds);
  	TAILQ_FOREACH(es, &agEventSource->sinks, sinks) {
  		switch (es->type) {
  		case AG_SINK_READ:
  			FD_SET(es->ident, &rdFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		case AG_SINK_WRITE:
  			FD_SET(es->ident, &wrFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		default:
  			break;
  		}
  	}
  #  ifdef AG_TIMERS
  	TAILQ_FOREACH(ob, &agTimerObjQ, tobjs) {
  		TAILQ_FOREACH(to, &ob->timers, pvt.timers) {
  			FD_SET(to->id, &rdFds);
  			if (to->id > nFds) { nFds = to->id; }
  		}
  	}
  #  endif
  	if (!TAILQ_EMPTY(&agEventSource->spinners)) {
  		timeo.tv_sec = 0;
  		timeo.tv_usec = 0;
  		pTimeo = &timeo;
  	} else {
  		pTimeo = NULL;
  	}
  	rv = select(nFds+1, &rdFds, &wrFds, NULL, pTimeo);
  	if (rv == -1) {
  		if (errno == EINTR) {
  			goto restart;
  		}
  		AG_SetError("select: %s", AG_Strerror(errno));
  		return (-1);
  	}
  
  #  ifdef AG_TIMERS
  	/* 1. Process timer expirations. */
  	AG_LockTiming();
  	for (ob = TAILQ_FIRST(&agTimerObjQ);
  	     ob != TAILQ_END(&agTimerObjQ);
  	     ob = obNext) {
  		obNext = TAILQ_NEXT(ob, tobjs);
  		AG_ObjectLock(ob);
  		for (to = TAILQ_FIRST(&ob->timers);
  		     to != TAILQ_END(&ob->timers);
  		     to = toNext) {
  			struct itimerspec its;
  			Uint32 rvt;
  
  			toNext = TAILQ_NEXT(to, pvt.timers);
  			if (!FD_ISSET(to->id, &rdFds)) {
  				continue;
  			}
  			rvt = to->fn(to, &to->fnEvent);
  			if (rvt > 0) {
  				its.it_value.tv_sec = rvt/1000;
  				its.it_value.tv_nsec = (rvt % 1000)*1000000L;
  				its.it_interval.tv_sec = 0;
  				its.it_interval.tv_nsec = 0L;
  				if (timerfd_settime(to->id, 0, &its, NULL) == -1) {
  					Verbose("timerfd_settime: %s\n", AG_Strerror(errno));
  					FD_CLR(to->id, &rdFds);
  					AG_DelTimer(ob, to);
  				}
  			} else {
  				FD_CLR(to->id, &rdFds);
  				AG_DelTimer(ob, to);
  			}
  		}
  		AG_ObjectUnlock(ob);
  	}
  	AG_UnlockTiming();
  #  endif /* AG_TIMERS */
  	
  	/* 2. Process I/O events. */
  	for (es = TAILQ_FIRST(&agEventSource->sinks);
  	     es != TAILQ_END(&agEventSource->sinks);
  	     es = esNext) {
  		esNext = TAILQ_NEXT(es, sinks);
  		switch (es->type) {
  		case AG_SINK_READ:
  			if (FD_ISSET(es->ident, &rdFds)) {
  				es->fn(es, &es->fnArgs);
  			}
  			break;
  		case AG_SINK_WRITE:
  			if (FD_ISSET(es->ident, &wrFds)) {
  				es->fn(es, &es->fnArgs);
  			}
  			break;
  		default:
  			break;
  		}
  	}
  	return (0);
  }
  
  #  ifdef AG_TIMERS
  /*
   * Add/remove a fd-based timer.
   */
  int
  AG_AddTimerTIMERFD(AG_Timer *to, Uint32 ival, int newTimer)
  {
  	struct itimerspec its;
  
  	if (newTimer) {
  		/* Create a timerfd. Store the file descriptor as ID. */
  		if ((to->id = timerfd_create(CLOCK_MONOTONIC, 0)) == -1) {
  			AG_SetError("timerfd_create: %s", AG_Strerror(errno));
  			return (-1);
  		}
  	}
  	its.it_value.tv_sec = ival/1000;
  	its.it_value.tv_nsec = (ival % 1000)*1000000L;
  	its.it_interval.tv_sec = 0;
  	its.it_interval.tv_nsec = 0L;
  	if (timerfd_settime(to->id, 0, &its, NULL) == -1) {
  		close(to->id);
  		AG_SetError("timerfd_settime: %s", AG_Strerror(errno));
  		return (-1);
  	}
  	to->ival = ival;
  	return (0);
  }
  void
  AG_DelTimerTIMERFD(AG_Timer *to)
  {
  #  ifdef AG_DEBUG
  	if (to->id == -1)
  		AG_FatalError("timerfd inconsistency");
  #  endif
  	close(to->id);
  }
  #  endif /* AG_TIMERS */
  # endif /* HAVE_TIMERFD */
  
  # if defined(HAVE_SELECT) && !defined(AG_THREADS)
  /*
   * Standard event sink using select(2) with timers implemented using the
   * select() timeout. Only available in single-threaded builds, since timers
   * cannot be added, restarted or removed from different threads with this
   * method.
   */
  int
  AG_EventSinkTIMEDSELECT(void)
  {
  	fd_set rdFds, wrFds;
  	int i, nFds, rv;
  	AG_EventSink *es;
  #  ifdef AG_TIMERS
  	AG_Object *ob, *obNext;
  	AG_Timer *to, *toNext;
  	struct timeval timeo, *pTimeo;
  	Uint32 t, tSoonest;
  #  endif
  
  restart:
  	nFds = 0;
  	FD_ZERO(&rdFds);
  	FD_ZERO(&wrFds);
  	TAILQ_FOREACH(es, &agEventSource->sinks, sinks) {
  		switch (es->type) {
  		case AG_SINK_READ:
  			FD_SET(es->ident, &rdFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		case AG_SINK_WRITE:
  			FD_SET(es->ident, &wrFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		}
  	}
  #  ifdef AG_TIMERS
  	if (!TAILQ_EMPTY(&agEventSource->spinners)) {
  		timeo.tv_sec = 0;
  		timeo.tv_usec = 0;
  	} else {
  		AG_LockTiming();
  		t = AG_GetTicks();
  		tSoonest = 0xfffffffe;
  		TAILQ_FOREACH(ob, &agTimerObjQ, tobjs) {
  			TAILQ_FOREACH(to, &ob->timers, pvt.timers) {
  				if ((to->tSched - t) < tSoonest)
  					tSoonest = (to->tSched - t);
  			}
  		}
  		timeo.tv_sec = tSoonest/1000;
  		timeo.tv_usec = (tSoonest % 1000)*1000;
  		AG_UnlockTiming();
  	}
  #  else /* !AG_TIMERS */
  	timeo.tv_sec = 0;
  	timeo.tv_usec = 0;
  #  endif /* AG_TIMERS */
  	rv = select(nFds+1, &rdFds, &wrFds, NULL, &timeo);
  	if (rv == -1) {
  		if (errno == EINTR) {
  			goto restart;
  		}
  		AG_SetError("select: %s", AG_Strerror(errno));
  		return (-1);
  	}
  #  ifdef AG_TIMERS
  	AG_LockTiming();
  	/* 1. Process timer expirations. */
  	AG_ProcessTimeouts(t);
  #  endif
  	if (rv > 0) {
  		/* 2. Process I/O events */
  		TAILQ_FOREACH(es, &agEventSource->sinks, sinks) {
  			switch (es->type) {
  			case AG_SINK_READ:
  				if (FD_ISSET(es->ident, &rdFds)) {
  					es->fn(es, &es->fnArgs);
  				}
  				break;
  			case AG_SINK_WRITE:
  				if (FD_ISSET(es->ident, &wrFds)) {
  					es->fn(es, &es->fnArgs);
  				}
  				break;
  			}
  		}
  	}
  #  ifdef AG_TIMERS
  	AG_UnlockTiming();
  #  endif
  	return (0);
  }
  # endif /* HAVE_SELECT and !AG_THREADS */
  
  # if defined(HAVE_SELECT) && defined(AG_THREADS)
  /*
   * Standard event sink using non-blocking select(2) with timers implemented
   * with a delay loop. This is inefficient, but on some platforms, it is the
   * only thread-safe option.
   */
  int
  AG_EventSinkSELECT(void)
  {
  	fd_set rdFds, wrFds;
  	int nFds, rv;
  	AG_EventSink *es;
  	struct timeval timeo;
  
  	nFds = 0;
  	FD_ZERO(&rdFds);
  	FD_ZERO(&wrFds);
  	
  	TAILQ_FOREACH(es, &agEventSource->sinks, sinks) {
  		switch (es->type) {
  		case AG_SINK_READ:
  			FD_SET(es->ident, &rdFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		case AG_SINK_WRITE:
  			FD_SET(es->ident, &wrFds);
  			if (es->ident > nFds) { nFds = es->ident; }
  			break;
  		default:
  			break;
  		}
  	}
  
  restart:
  	timeo.tv_sec = 0;
  	timeo.tv_usec = 0;
  	rv = select(nFds+1, &rdFds, &wrFds, NULL, &timeo);
  	if (rv == -1) {
  		if (errno == EINTR) {
  			goto restart;
  		}
  		AG_SetError("select: %s", AG_Strerror(errno));
  		return (-1);
  	}
  #  ifdef AG_TIMERS
  	AG_LockTiming();
  	/* 1. Process timer expirations. */
  	AG_ProcessTimeouts(AG_GetTicks());
  #  endif
  	if (rv > 0) {
  		/* 2. Process I/O events. */
  		TAILQ_FOREACH(es, &agEventSource->sinks, sinks) {
  			switch (es->type) {
  			case AG_SINK_READ:
  				if (FD_ISSET(es->ident, &rdFds)) {
  					es->fn(es, &es->fnArgs);
  				}
  				break;
  			case AG_SINK_WRITE:
  				if (FD_ISSET(es->ident, &wrFds)) {
  					es->fn(es, &es->fnArgs);
  				}
  				break;
  			default:
  				break;
  			}
  		}
  	}
  #  ifdef AG_TIMERS
  	AG_UnlockTiming();
  #  endif
  	if (TAILQ_EMPTY(&agEventSource->spinners)) {
  		AG_Delay(1);
  	}
  	return (0);
  }
  # endif /* HAVE_SELECT and AG_THREADS */
  
  /*
   * Fallback "spinning" event sink using a delay loop. This is inefficient,
   * but is the only option on some platforms.
   */
  int
  AG_EventSinkSPINNER(void)
  {
  # ifdef AG_TIMERS
  	AG_ProcessTimeouts(AG_GetTicks());
  # endif
  	AG_Delay(1);
  	return (0);
  }
  
  /*
   * Standard event loop routine. We loop over the event source and invoke
   * the related event sinks. AG_EventLoop() may be used outside of the
   * main thread (in which case, a new event source will be created).
   */
  int
  AG_EventLoop(void)
  {
  	AG_EventSource *src = AG_GetEventSource();
  	AG_EventSink *es, *esNext;
  	
  	TAILQ_FOREACH(es, &src->prologues, sinks) {
  		es->fn(es, &es->fnArgs);
  	}
  	for (;;) {
  		TAILQ_FOREACH(es, &src->spinners, sinks) {
  			es->fn(es, &es->fnArgs);
  		}
  		if (src->sinkFn() == -1) {
  			return (1);
  		}
  		for (es = TAILQ_FIRST(&src->epilogues);
  		     es != TAILQ_END(&src->epilogues);
  		     es = esNext) {
  			esNext = TAILQ_NEXT(es, sinks);
  			es->fn(es, &es->fnArgs);
  		}
  		if (src->breakReq)
  			break;
  	}
  	return (src->returnCode);
  }
  
  /* Request that we break out of AG_EventLoop(). */
  void
  AG_Terminate(int retCode)
  {
  	AG_EventSource *src = AG_GetEventSource();
  
  	src->breakReq = 1;
  	src->returnCode = retCode;
  }
  void
  AG_TerminateEv(AG_Event *ev)
  {
  	AG_EventSource *src = AG_GetEventSource();
  
  	if (ev->argc > 1 &&
  	    ev->argv[1].type == AG_VARIABLE_INT) {
  		src->returnCode = ev->argv[1].data.i;
  	} else {
  		src->returnCode = 0;
  	}
  	src->breakReq = 1;
  }
  #endif /* AG_EVENT_LOOP */
  