Merge branch 'fixed_eventloop' of https://github.com/neilcook/getdns into neilcook-fixed_eventloop

src/extension/default_eventloop.c

@@ -27,10 +27,36 @@
 
 #include "config.h"
 
+#ifndef USE_WINSOCK
+#include <poll.h>
+#endif
+#include <sys/resource.h>
 #include "extension/default_eventloop.h"
 #include "debug.h"
 #include "types-internal.h"
 
+_getdns_eventloop_info *find_event(_getdns_eventloop_info** events, int id)
+{
+	_getdns_eventloop_info* ev;
+
+	HASH_FIND_INT(*events, &id, ev);
+
+	return ev;
+}
+
+void add_event(_getdns_eventloop_info** events, int id, _getdns_eventloop_info* ev)
+{
+	DEBUG_SCHED("default_eventloop: add_event with id %d\n", id);
+	ev->id = id;
+	HASH_ADD_INT(*events, id, ev);
+}
+
+void delete_event(_getdns_eventloop_info** events, _getdns_eventloop_info* ev)
+{
+	DEBUG_SCHED("default_eventloop: delete_event with id %d\n", ev->id);
+	HASH_DEL(*events, ev);
+}
+
 static uint64_t get_now_plus(uint64_t amount)
 {
 	struct timeval tv;
@@ -53,15 +79,15 @@ default_eventloop_schedule(getdns_eventloop *loop,
 	_getdns_default_eventloop *default_loop  = (_getdns_default_eventloop *)loop;
 	size_t i;
 
-	DEBUG_SCHED( "%s(loop: %p, fd: %d, timeout: %"PRIu64", event: %p, FD_SETSIZE: %d)\n"
+	DEBUG_SCHED( "%s(loop: %p, fd: %d, timeout: %"PRIu64", event: %p, max_fds: %d)\n"
-	        , __FUNC__, (void *)loop, fd, timeout, (void *)event, FD_SETSIZE);
+	        , __FUNC__, (void *)loop, fd, timeout, (void *)event, default_loop->max_fds);
 
 	if (!loop || !event)
 		return GETDNS_RETURN_INVALID_PARAMETER;
 
-	if (fd >= (int)FD_SETSIZE) {
+	if (fd >= (int)default_loop->max_fds) {
-		DEBUG_SCHED( "ERROR: fd %d >= FD_SETSIZE: %d!\n"
+		DEBUG_SCHED( "ERROR: fd %d >= max_fds: %d!\n"
-		           , fd, FD_SETSIZE);
+		           , fd, default_loop->max_fds);
 		return GETDNS_RETURN_GENERIC_ERROR;
 	}
 	if (fd >= 0 && !(event->read_cb || event->write_cb)) {
@@ -70,23 +96,32 @@ default_eventloop_schedule(getdns_eventloop *loop,
 		fd = -1;
 	}
 	if (fd >= 0) {
+		_getdns_eventloop_info* fd_event = find_event(&default_loop->fd_events, fd);
 #if defined(SCHED_DEBUG) && SCHED_DEBUG
-		if (default_loop->fd_events[fd]) {
+		if (fd_event) {
-			if (default_loop->fd_events[fd] == event) {
+			if (fd_event->event == event) {
 				DEBUG_SCHED("WARNING: Event %p not cleared "
 				            "before being rescheduled!\n"
-				           , (void *)default_loop->fd_events[fd]);
+				           , (void *)fd_event->event);
 			} else {
 				DEBUG_SCHED("ERROR: A different event is "
 				            "already present at fd slot: %p!\n"
-				           , (void *)default_loop->fd_events[fd]);
+				           , (void *)fd_event->event);
 			}
 		}
 #endif
-		default_loop->fd_events[fd] = event;
+		/* cleanup the old event if it exists */
-		default_loop->fd_timeout_times[fd] = get_now_plus(timeout);
+		if (fd_event) {
-		event->ev = (void *)(intptr_t)(fd + 1);
+			delete_event(&default_loop->fd_events, fd_event);
-		DEBUG_SCHED( "scheduled read/write at %d\n", fd);
+			free(fd_event);
+		}
+		fd_event = calloc(1, sizeof(_getdns_eventloop_info));
+		fd_event->event = event;
+		fd_event->timeout_time = get_now_plus(timeout);
+		add_event(&default_loop->fd_events, fd, fd_event);
+		event->ev = (void *) (intptr_t) (fd + 1);
+
+		DEBUG_SCHED( "scheduled read/write at fd %d\n", fd);
 		return GETDNS_RETURN_GOOD;
 	}
 	if (!event->timeout_cb) {
@@ -101,12 +136,16 @@ default_eventloop_schedule(getdns_eventloop *loop,
 		DEBUG_SCHED("ERROR: timeout event with write_cb! Clearing.\n");
 		event->write_cb = NULL;
 	}
-	for (i = 0; i < MAX_TIMEOUTS; i++) {
+	for (i = 0; i < default_loop->max_timeouts; i++) {
-		if (default_loop->timeout_events[i] == NULL) {
+		_getdns_eventloop_info* timeout_event = NULL;
-			default_loop->timeout_events[i] = event;
+		if ((timeout_event = find_event(&default_loop->timeout_events, i)) == NULL) {
-			default_loop->timeout_times[i] = get_now_plus(timeout);		
+			timeout_event = calloc(1, sizeof(_getdns_eventloop_info));
-			event->ev = (void *)(intptr_t)(i + 1);
+			timeout_event->event = event;
-			DEBUG_SCHED( "scheduled timeout at %d\n", (int)i);
+			timeout_event->timeout_time = get_now_plus(timeout);
+			add_event(&default_loop->timeout_events, i, timeout_event);
+			event->ev = (void *) (intptr_t) (i + 1);
+
+			DEBUG_SCHED( "scheduled timeout at slot %d\n", (int)i);
 			return GETDNS_RETURN_GOOD;
 		}
 	}
@@ -126,25 +165,34 @@ default_eventloop_clear(getdns_eventloop *loop, getdns_eventloop_event *event)
 	DEBUG_SCHED( "%s(loop: %p, event: %p)\n", __FUNC__, (void *)loop, (void *)event);
 
 	i = (intptr_t)event->ev - 1;
-	if (i < 0 || i >= FD_SETSIZE) {
+	if (i < 0 || i > default_loop->max_fds) {
 		return GETDNS_RETURN_GENERIC_ERROR;
 	}
 	if (event->timeout_cb && !event->read_cb && !event->write_cb) {
+		_getdns_eventloop_info* timeout_event = find_event(&default_loop->timeout_events, i);
 #if defined(SCHED_DEBUG) && SCHED_DEBUG
-		if (default_loop->timeout_events[i] != event)
+		if (timeout_event && timeout_event->event != event)
 			DEBUG_SCHED( "ERROR: Different/wrong event present at "
 			             "timeout slot: %p!\n"
-			           , (void *)default_loop->timeout_events[i]);
+				     , (void *)timeout_event);
+
 #endif
-		default_loop->timeout_events[i] = NULL;
+		if (timeout_event) {
+			delete_event(&default_loop->timeout_events, timeout_event);
+			free(timeout_event);
+		}
 	} else {
+		_getdns_eventloop_info* fd_event = find_event(&default_loop->fd_events, i);
 #if defined(SCHED_DEBUG) && SCHED_DEBUG
-		if (default_loop->fd_events[i] != event)
+		if (fd_event && fd_event->event != event)
 			DEBUG_SCHED( "ERROR: Different/wrong event present at "
 			             "fd slot: %p!\n"
-			           , (void *)default_loop->fd_events[i]);
+			           , (void *)fd_event);
 #endif
-		default_loop->fd_events[i] = NULL;
+		if (fd_event) {
+			delete_event(&default_loop->fd_events, fd_event);
+			free(fd_event);
+		}
 	}
 	event->ev = NULL;
 	return GETDNS_RETURN_GOOD;
@@ -153,7 +201,9 @@ default_eventloop_clear(getdns_eventloop *loop, getdns_eventloop_event *event)
 static void
 default_eventloop_cleanup(getdns_eventloop *loop)
 {
-	(void)loop;
+	_getdns_default_eventloop *default_loop  = (_getdns_default_eventloop *)loop;
+	HASH_CLEAR(hh, default_loop->fd_events);
+	HASH_CLEAR(hh, default_loop->timeout_events);
 }
 
 static void
@@ -190,97 +240,127 @@ static void
 default_eventloop_run_once(getdns_eventloop *loop, int blocking)
 {
 	_getdns_default_eventloop *default_loop  = (_getdns_default_eventloop *)loop;
-
+	_getdns_eventloop_info *s, *tmp;
-	fd_set   readfds, writefds;
-	int      fd, max_fd = -1;
 	uint64_t now, timeout = TIMEOUT_FOREVER;
-	size_t   i;
+	size_t   i=0;
-	struct timeval tv;
+	int poll_timeout = 0;
+	struct pollfd* pfds = NULL;
+	unsigned int num_pfds = 0;
 
 	if (!loop)
 		return;
 
-	FD_ZERO(&readfds);
-	FD_ZERO(&writefds);
 	now = get_now_plus(0);
 
-	for (i = 0; i < MAX_TIMEOUTS; i++) {
+	HASH_ITER(hh, default_loop->timeout_events, s, tmp) {
-		if (!default_loop->timeout_events[i])
+		if (now > s->timeout_time)
-			continue;
+			default_timeout_cb(-1, s->event);
-		if (now > default_loop->timeout_times[i])
+		else if (s->timeout_time < timeout)
-			default_timeout_cb(-1, default_loop->timeout_events[i]);
+			timeout = s->timeout_time;
-		else if (default_loop->timeout_times[i] < timeout)
-			timeout = default_loop->timeout_times[i];
 	}
-	for (fd = 0; fd < (int)FD_SETSIZE; fd++) {
+	// first we count the number of fds that will be active
-		if (!default_loop->fd_events[fd])
+	HASH_ITER(hh, default_loop->fd_events, s, tmp) {
-			continue;
+		if (s->event->read_cb ||
-		if (default_loop->fd_events[fd]->read_cb)
+		    s->event->write_cb)
-			FD_SET(fd, &readfds);
+			num_pfds++;
-		if (default_loop->fd_events[fd]->write_cb)
+		if (s->timeout_time < timeout)
-			FD_SET(fd, &writefds);
+			timeout = s->timeout_time;
-		if (fd > max_fd)
-			max_fd = fd;
-		if (default_loop->fd_timeout_times[fd] < timeout)
-			timeout = default_loop->fd_timeout_times[fd];
 	}
-	if (max_fd == -1 && timeout == TIMEOUT_FOREVER)
+
+	if ((timeout == TIMEOUT_FOREVER) && (num_pfds == 0))
 		return;
 
-	if (! blocking || now > timeout) {
+	pfds = calloc(num_pfds, sizeof(struct pollfd));
-		tv.tv_sec = 0;
+	i = 0;
-		tv.tv_usec = 0;
+	HASH_ITER(hh, default_loop->fd_events, s, tmp) {
-	} else {
+		if (s->event->read_cb) {
-		tv.tv_sec  = (long)((timeout - now) / 1000000);
+			pfds[i].fd = s->id;
-		tv.tv_usec = (long)((timeout - now) % 1000000);
+			pfds[i].events |= POLLIN;
 		}	
-	if (select(max_fd + 1, &readfds, &writefds, NULL,
+		if (s->event->write_cb) {
-	    (timeout == TIMEOUT_FOREVER ? NULL : &tv)) < 0) {
+			pfds[i].fd = s->id;
-		perror("select() failed");
+			pfds[i].events |= POLLOUT;
+		}
+		i++;
+	}
+
+	if (timeout == TIMEOUT_FOREVER) {
+		poll_timeout = -1;
+	}
+	else if (! blocking || now > timeout) {
+		poll_timeout = 0;
+	} else {
+		poll_timeout = (timeout - now) * 1000; /* turn seconds into millseconds */
+	}
+#ifdef USE_WINSOCK
+	if (WSAPoll(pfds, num_pfds, poll_timeout) < 0) {
+#else	
+	if (poll(pfds, num_pfds, poll_timeout) < 0) {
+#endif
+		perror("poll() failed");
 		exit(EXIT_FAILURE);
 	}
 	now = get_now_plus(0);
-	for (fd = 0; fd < (int)FD_SETSIZE; fd++) {
+	for (i = 0; i < num_pfds; i++) {
-		if (default_loop->fd_events[fd] &&
+		int fd = pfds[i].fd;
-		    default_loop->fd_events[fd]->read_cb &&
+		_getdns_eventloop_info* fd_event = find_event(&default_loop->fd_events, fd);
-		    FD_ISSET(fd, &readfds))
+		if (fd_event && fd_event->event) { 
-			default_read_cb(fd, default_loop->fd_events[fd]);
+			getdns_eventloop_event* event = fd_event->event;
+			if (event->read_cb &&
+			    (pfds[i].revents & POLLIN))
+				default_read_cb(fd, event);
 
-		if (default_loop->fd_events[fd] &&
+			if (event->write_cb &&
-		    default_loop->fd_events[fd]->write_cb &&
+			    (pfds[i].revents & POLLOUT))
-		    FD_ISSET(fd, &writefds))
+				default_write_cb(fd, event);
-			default_write_cb(fd, default_loop->fd_events[fd]);
-
-		if (default_loop->fd_events[fd] &&
-		    default_loop->fd_events[fd]->timeout_cb &&
-		    now > default_loop->fd_timeout_times[fd])
-			default_timeout_cb(fd, default_loop->fd_events[fd]);
-
-		i = fd;
-		if (default_loop->timeout_events[i] &&
-		    default_loop->timeout_events[i]->timeout_cb &&
-		    now > default_loop->timeout_times[i])
-			default_timeout_cb(-1, default_loop->timeout_events[i]);
 		}
+	}
+	if (pfds)
+		free(pfds);
+	_getdns_eventloop_info* fd_timeout_cbs = NULL;
+	HASH_ITER(hh, default_loop->fd_events, s, tmp) {
+		if (s->event &&
+		    s->event->timeout_cb &&
+		    now > s->timeout_time)
+			add_event(&fd_timeout_cbs, s->id, s);
+	}
+	/* this is in case the timeout callback deletes the event
+	   and thus messes with the iteration */
+	HASH_ITER(hh, fd_timeout_cbs, s, tmp) {
+		int fd = s->id;
+		getdns_eventloop_event* event = s->event;
+		delete_event(&fd_timeout_cbs, s);
+		default_timeout_cb(fd, event);
+	}
+	_getdns_eventloop_info* timeout_timeout_cbs = NULL;
+	HASH_ITER(hh, default_loop->timeout_events, s, tmp) {
+		if (s->event &&
+		    s->event->timeout_cb &&
+		    now > s->timeout_time)
+			add_event(&timeout_timeout_cbs, s->id, s);
+	}
+	/* this is in case the timeout callback deletes the event
+	   and thus messes with the iteration */
+	HASH_ITER(hh, timeout_timeout_cbs, s, tmp) {
+		int fd = s->id;
+		getdns_eventloop_event* event = s->event;
+		delete_event(&timeout_timeout_cbs, s);
+		default_timeout_cb(fd, event);
+	}
+
 }
 
 static void
 default_eventloop_run(getdns_eventloop *loop)
 {
 	_getdns_default_eventloop *default_loop  = (_getdns_default_eventloop *)loop;
-	size_t        i;
 
 	if (!loop)
 		return;
 
-	i = 0;
+	/* keep going until all the events are cleared */
-	while (i < MAX_TIMEOUTS) {
+	while (default_loop->fd_events || default_loop->timeout_events) {
-		if (default_loop->fd_events[i] || default_loop->timeout_events[i]) {
 		default_eventloop_run_once(loop, 1);
-			i = 0;
-		} else {
-			i++;
-		}
 	}
 }
 
@@ -297,4 +377,14 @@ _getdns_default_eventloop_init(_getdns_default_eventloop *loop)
 
 	(void) memset(loop, 0, sizeof(_getdns_default_eventloop));
 	loop->loop.vmt = &default_eventloop_vmt;
+
+	struct rlimit rl;
+	if (getrlimit(RLIMIT_NOFILE, &rl) == 0) {
+		loop->max_fds = rl.rlim_cur;
+		loop->max_timeouts = loop->max_fds; /* this is somewhat arbitrary */
+	} else {
+		DEBUG_SCHED("ERROR: could not obtain RLIMIT_NOFILE from getrlimit()\n");
+		loop->max_fds = 0;
+		loop->max_timeouts = loop->max_fds;
+	}
 }

src/extension/default_eventloop.h

@@ -34,24 +34,25 @@
 #include "config.h"
 #include "getdns/getdns.h"
 #include "getdns/getdns_extra.h"
+#include "util/uthash.h"
 
-/* No more than select's capability queries can be outstanding,
+/* Eventloop based on poll */
- * The number of outstanding timeouts should be less or equal then
+
- * the number of outstanding queries, so MAX_TIMEOUTS equal to
+typedef struct _getdns_eventloop_info {
- * FD_SETSIZE should be safe.
+	int			id;
- */
+	getdns_eventloop_event *event;
-#define MAX_TIMEOUTS FD_SETSIZE
+	uint64_t                timeout_time;
+	UT_hash_handle		hh;
+} _getdns_eventloop_info;
 
-/* Eventloop based on select */
 typedef struct _getdns_default_eventloop {
 	getdns_eventloop        loop;
-	getdns_eventloop_event *fd_events[FD_SETSIZE];
+	unsigned int		max_fds;
-	uint64_t                fd_timeout_times[FD_SETSIZE];
+	unsigned int		max_timeouts;
-	getdns_eventloop_event *timeout_events[MAX_TIMEOUTS];
+	_getdns_eventloop_info  *fd_events;
-	uint64_t                timeout_times[MAX_TIMEOUTS];
+	_getdns_eventloop_info  *timeout_events;
 } _getdns_default_eventloop;
 
-
 void
 _getdns_default_eventloop_init(_getdns_default_eventloop *loop);