- Rework event handling system so we don't use a full thread per

   connection, but instead only use threads for active requests.

10 files changed, 2048 insertions, 806 deletions

diff --git a/ChangeLog b/ChangeLog
index 0d0912e..d9835db 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,7 @@
+0.9.4 ???
+ - Rework event handling system so we don't use a full thread per
+   connection, but instead only use threads for active requests.
+
 0.9.3 ???
  - Support sending access groups to mod_httpauth apache2x module.
  - Support retrieving LDAP access groups for users.
diff --git a/common/server-mainloop.c b/common/server-mainloop.c
new file mode 100644
index 0000000..c102b88
--- /dev/null
+++ b/common/server-mainloop.c
@@ -0,0 +1,583 @@
+/*
+ * Copyright (c) 2006, Stefan Walter
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ *	 * Redistributions of source code must retain the above
+ *	   copyright notice, this list of conditions and the
+ *	   following disclaimer.
+ *	 * 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.
+ *	 * The names of contributors to this software may not be
+ *	   used to endorse or promote products derived from this
+ *	   software without specific prior written permission.
+ *
+ * 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
+ * COPYRIGHT OWNER 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.
+ */
+
+#include "config.h"
+#include "server-mainloop.h"
+
+#include <sys/time.h>
+
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#ifdef HAVE_PTHREAD
+#include <pthread.h>
+#endif
+
+typedef struct _socket_callback {
+	int fd;
+	server_socket_callback callback;
+	void *arg;
+	int closed;
+
+	struct _socket_callback *next;
+} socket_callback;
+
+typedef struct _timer_callback {
+	struct timeval at;
+	struct timeval interval;
+	server_timer_callback callback;
+	void* arg;
+	int closed;
+
+	struct _timer_callback *next;
+} timer_callback;
+
+typedef struct _server_context {
+	int stopped;
+	fd_set read_fds;
+	fd_set write_fds;
+	int max_fd;
+	int revision;
+	socket_callback *callbacks;
+	timer_callback *timers;
+	int wakeup[2];
+#ifdef HAVE_PTHREAD
+	pthread_mutex_t mutex;
+#endif
+} server_context;
+
+#ifdef HAVE_PTHREAD
+#define LOCK() pthread_mutex_lock (&ctx.mutex)
+#define UNLOCK() pthread_mutex_unlock (&ctx.mutex)
+#else
+#define LOCK()
+#define UNLOCK()
+#endif
+
+/* Global context */
+static server_context ctx;
+
+static void
+timeval_add(struct timeval *t1, struct timeval *t2)
+{
+	assert (t1->tv_usec < 1000000);
+	assert (t2->tv_usec < 1000000);
+
+	t1->tv_sec += t2->tv_sec;
+	t1->tv_usec += t2->tv_usec;
+	if (t1->tv_usec >= 1000000) {
+		t1->tv_usec -= 1000000;
+		t1->tv_sec += 1;
+	}
+}
+
+static void
+timeval_subtract (struct timeval *t1, struct timeval *t2)
+{
+	assert (t1->tv_usec < 1000000);
+	assert (t2->tv_usec < 1000000);
+
+	t1->tv_sec -= t2->tv_sec;
+	if (t1->tv_usec < t2->tv_usec) {
+		t1->tv_usec += 1000000;
+		t1->tv_sec -= 1;
+	}
+	t1->tv_usec -= t2->tv_usec;
+}
+
+static int
+timeval_compare(struct timeval *t1, struct timeval *t2)
+{
+	assert (t1->tv_usec < 1000000);
+	assert (t2->tv_usec < 1000000);
+
+	if (t1->tv_sec > t2->tv_sec)
+		return 1;
+	else if (t1->tv_sec < t2->tv_sec)
+		return -1;
+	else {
+		if (t1->tv_usec > t2->tv_usec)
+			return 1;
+		else if (t1->tv_usec < t2->tv_usec)
+			return -1;
+		else
+			return 0;
+	}
+}
+
+#define timeval_empty(tv) \
+	((tv)->tv_sec == 0 && (tv)->tv_usec == 0)
+
+#define timeval_to_ms(tv) \
+	((((uint64_t)(tv).tv_sec) * 1000L) + (((uint64_t)(tv).tv_usec) / 1000L))
+
+#define timeval_dump(tv) \
+	(fprintf(stderr, "{ %d:%d }", (uint)((tv).tv_sec), (uint)((tv).tv_usec / 1000)))
+
+static int
+add_timer (int ms, int oneshot, server_timer_callback callback, void *arg)
+{
+	struct timeval interval;
+	timer_callback *cb;
+
+	assert (ms || oneshot);
+	assert (callback != NULL);
+
+	interval.tv_sec = ms / 1000;
+	interval.tv_usec = (ms % 1000) * 1000; /* into micro seconds */
+
+	cb = (timer_callback*)calloc (1, sizeof (*cb));
+	if(!cb) {
+		errno = ENOMEM;
+		return -1;
+	}
+
+	if (gettimeofday (&(cb->at), NULL) == -1) {
+		free(cb);
+		return -1;
+	}
+
+	timeval_add (&(cb->at), &interval);
+
+	if (oneshot)
+		memset (&(cb->interval), 0, sizeof (cb->interval));
+	else
+		memcpy (&(cb->interval), &interval, sizeof(cb->interval));
+
+	cb->callback = callback;
+	cb->arg = arg;
+
+	cb->next = ctx.timers;
+	ctx.timers = cb;
+	++ctx.revision;
+	server_wakeup ();
+
+	return 0;
+}
+
+static void
+remove_closed (void)
+{
+	timer_callback **tcb, *timcb;
+	socket_callback **scb, *sockcb;
+
+	for (tcb = &ctx.timers; *tcb; ) {
+		timcb = *tcb;
+		if (timcb->closed) {
+			*tcb = timcb->next;
+			free (timcb);
+			continue;
+		} else {
+			tcb = &timcb->next;
+		}
+	}
+
+	for (scb = &ctx.callbacks; *scb; ) {
+		sockcb = *scb;
+		if (sockcb->closed) {
+			*scb = sockcb->next;
+			free (sockcb);
+			continue;
+		} else {
+			scb = &sockcb->next;
+		}
+	}
+}
+
+int
+server_init (void)
+{
+	memset (&ctx, 0, sizeof (ctx));
+
+	if (pipe (ctx.wakeup) < 0)
+		return -1;
+
+	FD_ZERO (&ctx.read_fds);
+	FD_ZERO (&ctx.write_fds);
+
+	/* The wakeup descriptor */
+	FD_SET (ctx.wakeup[0], &ctx.read_fds);
+	fcntl (ctx.wakeup[0], F_SETFL, fcntl (ctx.wakeup[0], F_GETFL, 0) | O_NONBLOCK);
+
+	ctx.max_fd = ctx.wakeup[0] + 1;
+
+	ctx.stopped = 1;
+	ctx.callbacks = NULL;
+	ctx.timers = NULL;
+
+
+#ifdef HAVE_PTHREAD
+	{
+		int ret = pthread_mutex_init (&ctx.mutex, NULL);
+		if (ret != 0) {
+			errno = ret;
+			close (ctx.wakeup[0]);
+			close (ctx.wakeup[1]);
+			return -1;
+		}
+	}
+#endif
+
+	return 0;
+}
+
+void
+server_uninit (void)
+{
+	timer_callback *timcb;
+	timer_callback *timn;
+	socket_callback *sockcb;
+	socket_callback *sockn;
+
+	LOCK ();
+
+	for(timcb = ctx.timers; timcb; timcb = timn) {
+		timn = timcb->next;
+		free (timcb);
+	}
+
+	ctx.timers = NULL;
+
+	for(sockcb = ctx.callbacks; sockcb; sockcb = sockn) {
+		sockn = sockcb->next;
+		free (sockcb);
+	}
+
+	ctx.callbacks = NULL;
+
+	UNLOCK ();
+
+#ifdef HAVE_PTHREAD
+	pthread_mutex_destroy (&ctx.mutex);
+#endif
+}
+
+uint64_t
+server_get_time (void)
+{
+	struct timeval tv;
+	if (gettimeofday (&tv, NULL) == -1)
+		return 0L;
+	return timeval_to_ms (tv);
+}
+
+int
+server_run (void)
+{
+	char buffer[16];
+	struct timeval *timeout;
+	struct timeval tv, current;
+	timer_callback *timcb;
+	socket_callback *sockcb;
+	fd_set rfds, wfds;
+	int r, revision, max_fd;
+	int ret = 0;
+
+	/* For doing lock safe callbacks */
+	int type, fd;
+	server_socket_callback scb;
+	server_timer_callback tcb;
+	void *arg;
+
+	LOCK ();
+
+	/* No watches have been set */
+	assert (ctx.max_fd > -1);
+
+	ctx.stopped = 0;
+	ctx.revision = 0;
+
+	while (!ctx.stopped)  {
+
+		/* Drain the wakeup if necessary */
+		while (read (ctx.wakeup[0], buffer, sizeof (buffer)) > 0);
+
+		/* Watch for the various fds */
+		memcpy (&rfds, &ctx.read_fds, sizeof (rfds));
+		memcpy (&wfds, &ctx.write_fds, sizeof (wfds));
+
+		/* Prepare for timers */
+		timeout = NULL;
+		if (gettimeofday (&current, NULL) == -1) {
+			ret = -1;
+			break;
+		}
+
+		/* Cycle through timers */
+		for (timcb = ctx.timers; timcb; timcb = timcb->next) {
+			assert (timcb->callback);
+
+			if (timcb->closed)
+				continue;
+
+			/* Call any timers that have already passed */
+			if (timeval_compare (&current, &timcb->at) >= 0) {
+
+				/* Do a lock safe callback */
+				tcb = timcb->callback;
+				arg = timcb->arg;
+
+				/* Convert to milliseconds, and make the call */
+				UNLOCK ();
+				r = (tcb) (timeval_to_ms (current), arg);
+				LOCK ();
+
+				/* Reset timer if so desired */
+				if (r == 1 && !timeval_empty (&timcb->interval)) {
+					timeval_add (&timcb->at, &timcb->interval);
+
+					/* If the time has already passed, just use current time */
+					if (timeval_compare (&(timcb->at), &current) <= 0)
+						memcpy (&(timcb->at), &current, sizeof (timcb->at));
+
+				/* Otherwise remove it. Either one shot, or returned 0 */
+				} else {
+					timcb->closed = 1;
+					continue;
+				}
+			}
+
+			/* Get soonest timer */
+			if (!timeout || timeval_compare (&timcb->at, timeout) < 0)
+				timeout = &timcb->at;
+		}
+
+		/* Convert to an offset */
+		if (timeout) {
+			memcpy (&tv, timeout, sizeof (tv));
+			timeout = &tv;
+			timeval_subtract (timeout, &current);
+		}
+
+		if (ctx.stopped)
+			continue;
+
+		remove_closed ();
+
+		/* No callbacks? No use continuing */
+		if (!ctx.callbacks && !ctx.timers)
+			break;
+
+		/* fprintf(stderr, "selecting with timeout: ");
+		   timeval_dump(timeout);
+		   fprintf(stderr, "\n"); */
+
+		/* We can see if changes happen while in select */
+		revision = ctx.revision;
+		max_fd = ctx.max_fd;
+
+		UNLOCK ();
+		r = select (max_fd, &rfds, &wfds, NULL, timeout);
+		LOCK ();
+
+		if (r < 0) {
+			/* Interrupted so try again, and possibly exit */
+			if (errno == EINTR)
+				continue;
+
+			if (errno == EBADF) {
+				if (ctx.revision != revision)
+					continue;
+				assert (0 && "bad file descriptor being watched");
+			}
+
+			/* Programmer errors */
+			assert (errno != EINVAL);
+			ret = -1;
+			break;
+		}
+
+		/* Timeout, just jump to timeout processing */
+		if(r == 0)
+			continue;
+
+		/* Process ready file descriptors */
+		for (sockcb = ctx.callbacks; sockcb; sockcb = sockcb->next) {
+			assert (sockcb->fd != -1);
+			if (sockcb->closed)
+				continue;
+
+			/* Call any that are set */
+			type = 0;
+			if (FD_ISSET (sockcb->fd, &rfds))
+				type = SERVER_READ;
+			if (FD_ISSET (sockcb->fd, &wfds))
+				type = SERVER_WRITE;
+
+			/* Lock safe callback */
+			if (type != 0) {
+				scb = sockcb->callback;
+				fd = sockcb->fd;
+				arg = sockcb->arg;
+
+				UNLOCK ();
+					(scb) (fd, type, arg);
+				LOCK ();
+			}
+		}
+	}
+
+	UNLOCK ();
+	return ret;
+}
+
+void
+server_wakeup (void)
+{
+	/*
+	 * This can be called from within arbitrary
+	 * locks, so be careful.
+	 */
+	int r = write (ctx.wakeup[1], "a", 1);
+	fprintf (stderr, "wakeup: %d\n", r);
+}
+
+void
+server_stop (void)
+{
+	/*
+	 * This could be called from a signal handler,
+	 * so it's not worth trying to lock.
+	 */
+	ctx.stopped = 1;
+	server_wakeup ();
+}
+
+int
+server_stopped (void)
+{
+	return ctx.stopped;
+}
+
+int
+server_watch (int fd, int type, server_socket_callback callback, void *arg)
+{
+	socket_callback *cb;
+	assert (type != 0);
+	assert (fd != -1);
+	assert (callback != NULL);
+
+	cb = (socket_callback*)calloc (sizeof(*cb), 1);
+	if(!cb) {
+		errno = ENOMEM;
+		return -1;
+	}
+
+	LOCK ();
+
+	cb->fd = fd;
+	cb->callback = callback;
+	cb->arg = arg;
+
+	cb->next = ctx.callbacks;
+	ctx.callbacks = cb;
+
+	if (type & SERVER_READ)
+		FD_SET (fd, &ctx.read_fds);
+	if (type & SERVER_WRITE)
+		FD_SET (fd, &ctx.write_fds);
+
+	if (fd >= ctx.max_fd)
+		ctx.max_fd = fd + 1;
+
+	++ctx.revision;
+	server_wakeup ();
+
+	UNLOCK ();
+
+	return 0;
+}
+
+void
+server_unwatch (int fd)
+{
+	socket_callback* cb;
+
+	assert (fd != -1);
+
+	LOCK ();
+
+	FD_CLR (fd, &ctx.read_fds);
+	FD_CLR (fd, &ctx.write_fds);
+
+	if (!ctx.callbacks)
+		goto done;
+
+	/* First in list */;
+	if (ctx.callbacks->fd == fd) {
+		ctx.callbacks->closed = 1;
+		goto done;
+	}
+
+	/* One ahead processing of rest */
+	for (cb = ctx.callbacks; cb->next; cb = cb->next) {
+		if (cb->next->fd == fd) {
+			cb->next->closed = 1;
+			goto done;
+		}
+	}
+
+done:
+	server_wakeup ();
+	++ctx.revision;
+	UNLOCK ();
+}
+
+int
+server_timer (int ms, server_timer_callback callback, void *arg)
+{
+	int ret;
+
+	LOCK ();
+	ret = add_timer (ms, 0, callback, arg);
+	UNLOCK ();
+
+	return ret;
+}
+
+int
+server_oneshot (int ms, server_timer_callback callback, void *arg)
+{
+	int ret;
+
+	LOCK ();
+	ret = add_timer (ms, 1, callback, arg);
+	UNLOCK ();
+
+	return ret;
+}
diff --git a/common/server-mainloop.h b/common/server-mainloop.h
new file mode 100644
index 0000000..a749653
--- /dev/null
+++ b/common/server-mainloop.h
@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2006, Stefan Walter
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ *     * Redistributions of source code must retain the above
+ *       copyright notice, this list of conditions and the
+ *       following disclaimer.
+ *     * 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.
+ *     * The names of contributors to this software may not be
+ *       used to endorse or promote products derived from this
+ *       software without specific prior written permission.
+ *
+ * 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
+ * COPYRIGHT OWNER 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.
+ */
+
+#ifndef __SERVER_MAINLOOP_H__
+#define __SERVER_MAINLOOP_H__
+
+#include <stdint.h>
+
+#define SERVER_READ       0x01
+#define SERVER_WRITE      0x02
+
+typedef void (*server_socket_callback) (int fd, int type, void* arg);
+typedef int (*server_timer_callback) (uint64_t when, void* arg);
+
+int     server_init (void);
+void    server_uninit (void);
+int     server_run (void);
+void    server_stop (void);
+int     server_stopped (void);
+int     server_watch (int fd, int type, server_socket_callback callback, void* arg);
+void    server_unwatch (int fd);
+void    server_wakeup (void);
+int     server_timer (int length, server_timer_callback callback, void* arg);
+int     server_oneshot (int length, server_timer_callback callback, void* arg);
+uint64_t server_get_time (void);
+
+#endif /* __SERVER_MAINLOOP_H__ */
diff --git a/common/tpool.c b/common/tpool.c
new file mode 100644
index 0000000..f980b86
--- /dev/null
+++ b/common/tpool.c
@@ -0,0 +1,398 @@
+/*
+ * HttpAuth
+ *
+ * Copyright (C) 2008 Stefan Walter
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the
+ * Free Software Foundation, Inc.,
+ * 59 Temple Place, Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/*
+ * -------------------------------------------------------------------------
+ * Ideas from tpool.c - htun thread pool functions
+ * Copyright (C) 2002 Moshe Jacobson <moshe@runslinux.net>,
+ *                    Ola Nordström <ola@triblock.com>
+ * -------------------------------------------------------------------------
+ */
+
+/*
+ * most of the theory and implementation of the thread pool was taken
+ * from the o'reilly pthreads programming book.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <errno.h>
+#include <err.h>
+#include <string.h>
+#include <assert.h>
+#include <signal.h>
+
+#define __USE_GNU
+#include <pthread.h>
+
+#include "tpool.h"
+
+typedef struct _tpool_work {
+	void (*handler_routine)(void*);
+	void *arg;
+	struct _tpool_work *next;
+} tpool_work;
+
+typedef struct _tpool_thread {
+	pthread_t handle;
+	int is_active;
+	struct _tpool_thread *next;
+} tpool_thread;
+
+static int tpool_max_threads = 0;
+static int tpool_min_threads = 0;
+static int tpool_running_threads = 0;
+static int tpool_idle_threads = 0;
+static int tpool_zombie_threads = 0;
+static tpool_thread *tpool_thread_list = NULL;
+
+static int tpool_max_queue_size = 0;
+static int tpool_cur_queue_size = 0;
+static tpool_work *tpool_queue_head = NULL;
+static tpool_work *tpool_queue_tail = NULL;
+static pthread_mutex_t tpool_queue_lock;
+static pthread_cond_t tpool_queue_not_full;
+static pthread_cond_t tpool_queue_not_empty;
+static pthread_cond_t tpool_queue_empty;
+
+static int tpool_queue_closed = 0;
+static int tpool_shutdown = 0;
+static int tpool_initialized = 0;
+
+#define PTHREQ(x) do { \
+	int r = (x); \
+	if (r != 0) { \
+		fprintf (stderr, "fatal: '%s' failed: %s", #x, strerror (r)); \
+		assert (0 && "" #x); \
+	} } while (0)
+
+static void*
+tpool_thread_func (void *data)
+{
+	tpool_thread *thread = (tpool_thread*)data;
+	tpool_work *work;
+	int stop = 0;
+	sigset_t set;
+
+	/* We handle these signals on the main thread */
+	sigemptyset (&set);
+	sigaddset (&set, SIGINT);
+	sigaddset (&set, SIGTERM);
+	pthread_sigmask (SIG_BLOCK, &set, NULL);
+
+	assert (thread);
+	assert (tpool_initialized);
+
+	PTHREQ (pthread_mutex_lock (&tpool_queue_lock));
+
+	assert (thread->is_active);
+	++tpool_running_threads;
+	++tpool_idle_threads;
+
+	while (!tpool_shutdown && !stop) {
+
+		/* No work to do? */
+		while (tpool_cur_queue_size == 0 && !tpool_shutdown && !stop) {
+
+			/* Too many idle threads? */
+			if (tpool_running_threads > tpool_min_threads) {
+				stop = 1;
+				break;
+			}
+
+			/*
+			 * Sleep until there is work, while asleep the queue_lock
+			 * is relinquished
+			 */
+			PTHREQ (pthread_cond_wait (&tpool_queue_not_empty, &tpool_queue_lock));
+		}
+
+		/* Are we shutting down? */
+		if (stop || tpool_shutdown)
+			break;
+
+		/* Pop the work off the queue */
+		work = tpool_queue_head;
+		tpool_cur_queue_size--;
+		if (tpool_cur_queue_size == 0)
+			tpool_queue_head = tpool_queue_tail = NULL;
+		else
+			tpool_queue_head = work->next;
+
+		/* Broadcast queue state to anyone who's waiting */
+		if (tpool_cur_queue_size < tpool_max_queue_size)
+			PTHREQ (pthread_cond_broadcast (&tpool_queue_not_full));
+		if (tpool_cur_queue_size == 0)
+			PTHREQ (pthread_cond_signal (&tpool_queue_empty));
+
+		--tpool_idle_threads;
+		assert (tpool_idle_threads >= 0);
+
+		/* Unlock the queue lock while processing */
+		PTHREQ (pthread_mutex_unlock (&tpool_queue_lock));
+
+			/* Perform the work */
+			(work->handler_routine) (work->arg);
+			free (work);
+
+		PTHREQ (pthread_mutex_lock (&tpool_queue_lock));
+
+		++tpool_idle_threads;
+	}
+
+	--tpool_idle_threads;
+	--tpool_running_threads;
+	++tpool_zombie_threads;
+	thread->is_active = 0;
+	assert (tpool_running_threads >= 0);
+	assert (tpool_idle_threads >= 0);
+
+	PTHREQ (pthread_mutex_unlock (&tpool_queue_lock));
+
+	/* So we can double check the result code */
+	return thread;
+}
+
+static void
+free_pool_stuff (void)
+{
+	tpool_work *work;
+
+	if (!tpool_initialized)
+		return;
+
+	/* No threads should be running at this point */
+	assert (!tpool_thread_list);
+
+	/* Clean up memory */
+	while (tpool_queue_head != NULL) {
+		work = tpool_queue_head;
+		tpool_queue_head = tpool_queue_head->next;
+		free (work);
+	}
+
+	pthread_cond_destroy (&tpool_queue_not_full);
+	pthread_cond_destroy (&tpool_queue_not_empty);
+	pthread_cond_destroy (&tpool_queue_empty);
+	pthread_mutex_destroy (&tpool_queue_lock);
+
+	tpool_initialized = 0;
+}
+
+int
+tpool_init (int max_threads, int min_threads, int max_queued)
+{
+	assert (!tpool_initialized);
+
+	/* set the desired thread pool values */
+	tpool_max_threads = max_threads;
+	tpool_running_threads = 0;
+	tpool_idle_threads = 0;
+	tpool_zombie_threads = 0;
+	tpool_min_threads = min_threads;
+
+	/* initialize the work queue */
+	tpool_max_queue_size = max_queued;
+	tpool_cur_queue_size = 0;
+	tpool_queue_head = NULL;
+	tpool_queue_tail = NULL;
+	tpool_queue_closed = 0;
+	tpool_shutdown = 0;
+
+	/* create the mutexs and cond vars */
+	PTHREQ (pthread_mutex_init (&tpool_queue_lock, NULL));
+	PTHREQ (pthread_cond_init (&tpool_queue_not_empty, NULL));
+	PTHREQ (pthread_cond_init (&tpool_queue_not_full, NULL));
+	PTHREQ (pthread_cond_init (&tpool_queue_empty, NULL));
+
+	tpool_initialized = 1;
+	return 0;
+}
+
+static void
+join_pool_threads (int all)
+{
+	tpool_thread *thread, **spot;
+	void *exit_code;
+
+	/* Should be called from within the lock */
+
+	for (spot = &tpool_thread_list, thread = *spot;
+	     thread && (all || tpool_zombie_threads > 0); ) {
+
+		assert (thread->handle);
+
+		if (all || !thread->is_active) {
+			if (!thread->is_active)
+				--tpool_zombie_threads;
+
+			/* Join the thread */
+			PTHREQ (pthread_join (thread->handle, &exit_code));
+			assert (exit_code == thread);
+
+			/* Remove and get the next in list */
+			*spot = thread->next;
+			free (thread);
+
+		} else {
+			/* Just get the next in list */
+			spot = &thread->next;
+		}
+
+		thread = *spot;
+	}
+}
+
+int
+tpool_add_work (void (*routine)(void*), void *arg)
+{
+	tpool_work *work;
+	tpool_thread *thread;
+	int r, ret = -1;
+
+	assert (tpool_initialized);
+
+	PTHREQ (pthread_mutex_lock (&tpool_queue_lock));
+
+	/* Join any old threads */
+	join_pool_threads (0);
+
+	/* Should have found all of these above */
+	assert (tpool_zombie_threads == 0);
+
+	/* Are we closed for business? */
+	if (tpool_shutdown || tpool_queue_closed) {
+		errno = ECANCELED;
+		goto done;
+	}
+
+	/* No idle threads to run this work? */
+	if (!tpool_idle_threads) {
+
+		/* See if we can start another thread */
+		if (tpool_max_threads <= 0 ||
+		    tpool_running_threads < tpool_max_threads) {
+
+			/* The thread context */
+			thread = calloc (1, sizeof (tpool_thread));
+			if (thread == NULL) {
+				errno = ENOMEM;
+				goto done;
+			}
+
+			/* Start the thread */
+			thread->is_active = 1;
+			r = pthread_create (&thread->handle, NULL, tpool_thread_func, (void*)thread);
+			if (r != 0) {
+				errno = r;
+				free (thread);
+				goto done;
+			}
+
+			/* Add the new thread to the list */
+			thread->next = tpool_thread_list;
+			tpool_thread_list = thread;
+		}
+	}
+
+	/*
+	 * If the queue is full, yield this thread, and see if
+	 * another processes something out of the queue.
+	 */
+	if (tpool_max_queue_size > 0 && tpool_cur_queue_size >= tpool_max_queue_size) {
+		PTHREQ (pthread_mutex_unlock (&tpool_queue_lock));
+		pthread_yield();
+		PTHREQ (pthread_mutex_lock (&tpool_queue_lock));
+	}
+
+	/* Are we closed for business? */
+	if (tpool_shutdown || tpool_queue_closed) {
+		errno = ECANCELED;
+		goto done;
+	}
+
+	/* Is the queue still full? */
+	if (tpool_max_queue_size > 0 && tpool_cur_queue_size >= tpool_max_queue_size) {
+		errno = EAGAIN;
+		goto done;
+	}
+
+	/* Allocate the work structure */
+	work = calloc (1, sizeof (tpool_work));
+	if (work == NULL) {
+		errno = ENOMEM;
+		goto done;
+	}
+
+	work->handler_routine = routine;
+	work->arg = arg;
+	work->next = NULL;
+
+	if (tpool_cur_queue_size == 0) {
+		tpool_queue_tail = tpool_queue_head = work;
+	} else {
+		tpool_queue_tail->next = work;
+		tpool_queue_tail = work;
+	}
+
+	PTHREQ (pthread_cond_broadcast (&tpool_queue_not_empty));
+	tpool_cur_queue_size++;
+	ret = 0;
+
+done:
+	PTHREQ (pthread_mutex_unlock (&tpool_queue_lock));
+	return ret;
+}
+
+int
+tpool_destroy (int finish)
+{
+	assert (tpool_initialized);
+
+	PTHREQ (pthread_mutex_lock (&tpool_queue_lock));
+
+	/* Can't have two places calling shutdown */
+	assert (!tpool_queue_closed && !tpool_shutdown);
+
+	/* close the queue to any new work */
+	tpool_queue_closed = 1;
+
+	/* if the finish flag is set, drain the queue */
+	if (finish) {
+		while (tpool_cur_queue_size != 0)
+			PTHREQ (pthread_cond_wait (&tpool_queue_empty, &tpool_queue_lock));
+	}
+
+	/* Set the shutdown flag */
+	tpool_shutdown = 1;
+
+	PTHREQ (pthread_mutex_unlock (&tpool_queue_lock));
+
+	/* wake up all workers to rechedk the shutdown flag */
+	PTHREQ (pthread_cond_broadcast (&tpool_queue_not_empty));
+	PTHREQ (pthread_cond_broadcast (&tpool_queue_not_full));
+
+	join_pool_threads (1);
+	free_pool_stuff ();
+
+	return 0;
+}
+
diff --git a/common/tpool.h b/common/tpool.h
new file mode 100644
index 0000000..ac5a3f4
--- /dev/null
+++ b/common/tpool.h
@@ -0,0 +1,36 @@
+/*
+ * HttpAuth
+ *
+ * Copyright (C) 2008 Stefan Walter
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the
+ * Free Software Foundation, Inc.,
+ * 59 Temple Place, Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifndef _TPOOL_H_
+#define _TPOOL_H_
+
+typedef void (*tpool_routine) (void*);
+
+/* returns -1 if pool creation failed, 0 if succeeded */
+int tpool_init (int max_threads, int min_threads, int max_queued);
+
+/* returns -1 if failed (ie busy, closing etc...), 0 if succeeded */
+int tpool_add_work (tpool_routine routine, void *arg);
+
+/* cleanup and close. Must only be called once. Specify |finish| to drain queue */
+int tpool_destroy (int finish);
+
+#endif /* _TPOOL_H_ */
diff --git a/configure.in b/configure.in
index d57045e..513e325 100644
--- a/configure.in
+++ b/configure.in
@@ -36,8 +36,8 @@ dnl  Stef Walter <stef@memberwebs.com>
 dnl 
 
 dnl Process this file with autoconf to produce a configure script.
-AC_INIT(httpauth, 0.9.2, stef@memberwebs.com)
-AM_INIT_AUTOMAKE(httpauth, 0.9.2)
+AC_INIT(httpauth, 0.9.3, stef@memberwebs.com)
+AM_INIT_AUTOMAKE(httpauth, 0.9.3)
 
 LDFLAGS="$LDFLAGS -L/usr/local/lib"
 CFLAGS="$CFLAGS -I/usr/local/include -g -O0"
@@ -201,7 +201,7 @@ AC_ARG_ENABLE(debug,
 	    [Compile binaries in debug mode]))
 
 if test "$enable_debug" = "yes"; then
-  	CFLAGS="$CFLAGS -g -O0 -Wall -Werror"
+  	CFLAGS="$CFLAGS -g -O0 -Wall"
   	AC_DEFINE_UNQUOTED(_DEBUG, 1, [In debug mode])
   	echo "enabling debug compile mode"
 fi
diff --git a/daemon/Makefile.am b/daemon/Makefile.am
index 2e7be2f..b4b9f0c 100644
--- a/daemon/Makefile.am
+++ b/daemon/Makefile.am
@@ -28,10 +28,11 @@ EXTRA_SRC += $(MYSQL_SOURCES)
 endif
 
 httpauthd_SOURCES = httpauthd.c httpauthd.h usuals.h bd.h bd.c misc.c basic.h basic.c \
-    digest.h digest.c defaults.h simple.c dummy.c \
+    digest.h digest.c defaults.h simple.c dummy.c request.c \
     ../common/compat.h ../common/compat.c ../common/buffer.h ../common/buffer.c \
     ../common/hash.h ../common/hash.c ../common/md5.h ../common/md5.c \
     ../common/sha1.h ../common/sha1.c ../common/sock-any.c ../common/sock-any.h \
+    ../common/tpool.c ../common/tpool.h ../common/server-mainloop.c ../common/server-mainloop.h \
     ../common/stringx.c ../common/stringx.h $(EXTRA_SRC)
 
 httpauthd_CFLAGS = -D_THREAD_SAFE -pthread -DLinux \
diff --git a/daemon/httpauthd.c b/daemon/httpauthd.c
index 1adf8db..b75c116 100644
--- a/daemon/httpauthd.c
+++ b/daemon/httpauthd.c
@@ -37,14 +37,12 @@
 #include "usuals.h"
 #include "httpauthd.h"
 #include "defaults.h"
-#include "sock-any.h"
-#include "stringx.h"
 
-/*
- * This shouldn't be used by handlers,
- * they should return HA_FAILED instead.
- */
-#define HA_SERVER_ERROR    500
+#include "common/hash.h"
+#include "common/server-mainloop.h"
+#include "common/sock-any.h"
+#include "common/stringx.h"
+#include "common/tpool.h"
 
 /* -----------------------------------------------------------------------
  * Handlers Registered Here
@@ -88,36 +86,6 @@ httpauth_loaded_t* g_handlers = NULL;
 
 extern int pthread_mutexattr_settype (pthread_mutexattr_t *attr, int kind);
 
-/* -----------------------------------------------------------------------
- * Structures and Constants
- */
-
-/* A command definition. Used in parsing */
-typedef struct httpauth_command
-{
-    const char* name;
-    int code;
-    int word_args;          /* Arguments to be parsed as words */
-    int rest_arg;           /* Parse remainder as one arg? */
-    const char** headers;   /* Headers needed/allowed */
-}
-httpauth_command_t;
-
-/* The various valid headers for the auth command */
-const char* kAuthHeaders[] =
-{
-    "Authorization",
-    NULL
-};
-
-/* The command definitions */
-const httpauth_command_t kCommands[] =
-{
-    { "auth", REQTYPE_AUTH, 3, 0, kAuthHeaders },
-    { "set",  REQTYPE_SET,  1, 1, NULL         },
-    { "quit", REQTYPE_QUIT, 0, 0, NULL         },
-    { NULL,   -1,           0, 0, NULL         }
-};
 
 typedef struct httpauth_thread
 {
@@ -132,25 +100,23 @@ httpauth_thread_t;
 
 #define DEFAULT_CONFIG  CONF_PREFIX "/httpauthd.conf"
 #define DEFAULT_SOCKET  "/var/run/httpauthd.sock"
-#define DEFAULT_MAXTHREADS  32
+#define DEFAULT_MINTHREADS  8
+#define DEFAULT_MAXTHREADS  128
 
 
 /* -----------------------------------------------------------------------
  * Globals
  */
 
+int g_debuglevel = LOG_ERR;                      /* what gets logged to console */
 int g_daemonized = 0;                     /* Currently running as a daemon */
-int g_console = 0;                        /* debug mode read write from console */
-int g_debuglevel = LOG_ERR;               /* what gets logged to console */
-const char* g_socket = DEFAULT_SOCKET;    /* The socket to communicate on */
-int g_maxthreads = DEFAULT_MAXTHREADS;    /* The maximum number of threads */
-unsigned int g_unique = 0x10000;          /* A unique identifier (incremented) */
 
-/* For main loop and signal handlers */
-int g_quit = 0;
-
-/* The main thread */
-pthread_t g_mainthread;
+static int g_console = 0;                        /* debug mode read write from console */
+static const char* g_socket = DEFAULT_SOCKET;    /* The socket to communicate on */
+static int g_maxthreads = DEFAULT_MAXTHREADS;    /* The maximum number of threads */
+static int g_minthreads = DEFAULT_MINTHREADS;    /* The maximum number of threads */
+static unsigned int g_unique = 0x10000;          /* A unique identifier (incremented) */
+static hsh_t *g_requests = NULL;                 /* All the watched (ie: active) requests */
 
 /* The main mutex */
 pthread_mutex_t g_mutex;
@@ -162,9 +128,8 @@ pthread_mutexattr_t g_mutexattr;
 
 static int usage();
 static void writepid(const char* pid);
-static void* httpauth_thread(void* arg);
-static int httpauth_processor(int ifd, int ofd);
-static int httpauth_respond(ha_request_t* rq, int ofd, int scode, int ccode, const char* msg);
+static void accept_handler (int fd, int type, void *arg);
+static void close_all (void);
 static int config_parse(const char* file, ha_buffer_t* buf);
 static void on_quit(int signal);
 
@@ -176,7 +141,6 @@ int main(int argc, char* argv[])
 {
     const char* conf = DEFAULT_CONFIG;
     const char* pidfile = NULL;
-    httpauth_thread_t* threads = NULL;
     httpauth_loaded_t* h;
     struct sockaddr_any sany;
     int daemonize = 1;
@@ -184,9 +148,6 @@ int main(int argc, char* argv[])
     int r, i, sock;
     int ch = 0;
 
-    /* Keep note of the main thread */
-    g_mainthread = pthread_self();
-
     /* Create the main mutex */
     if(pthread_mutexattr_init(&g_mutexattr) != 0 ||
        pthread_mutexattr_settype(&g_mutexattr, HA_MUTEX_TYPE) ||
@@ -261,14 +222,13 @@ int main(int argc, char* argv[])
     /* Parse the configuration */
     config_parse(conf, &cbuf);
 
+    /* A hash table id -> request */
+    g_requests = hsh_create (sizeof (int));
+    if (g_requests == NULL)
+        err(1, "out of memory");
 
     if(!g_console)
     {
-        /* Create the thread buffers */
-        threads = (httpauth_thread_t*)calloc(g_maxthreads, sizeof(httpauth_thread_t));
-        if(!threads)
-            errx(1, "out of memory");
-
         /* Get the socket type */
         if(sock_any_pton(g_socket, &sany, SANY_OPT_DEFANY | SANY_OPT_DEFPORT(DEFAULT_PORT)) == -1)
             errx(1, "invalid socket name or ip: %s", g_socket);
@@ -310,7 +270,8 @@ int main(int argc, char* argv[])
     if(g_console)
     {
         ha_messagex(NULL, LOG_DEBUG, "processing from console");
-        r = httpauth_processor(0, 1);
+        ha_request_loop(0, 1);
+        r = 0;
         goto finally;
     }
 
@@ -348,100 +309,31 @@ int main(int argc, char* argv[])
 
         ha_messagex(NULL, LOG_DEBUG, "accepting connections");
 
-        /* Now loop and accept the connections */
-        while(!g_quit)
-        {
-            int fd;
-
-            fd = accept(sock, NULL, NULL);
-            if(fd == -1)
-            {
-                switch(errno)
-                {
-                case EINTR:
-                case EAGAIN:
-                    break;
-
-                case ECONNABORTED:
-                    ha_message(NULL, LOG_ERR, "couldn't accept a connection");
-                    break;
-
-                default:
-                    ha_message(NULL, LOG_ERR, "couldn't accept a connection");
-                    g_quit = 1;
-                    break;
-                };
-
-                if(g_quit)
-                    break;
-
-                continue;
-            }
-
-            memset(&sany, 0, sizeof(sany));
-            SANY_LEN(sany) = sizeof(sany);
-
-            /* Look for thread and also clean up others */
-            for(i = 0; i < g_maxthreads; i++)
-            {
-                /* Clean up quit threads */
-                if(threads[i].tid != 0)
-                {
-                    if(threads[i].fd == -1)
-                    {
-                        ha_messagex(NULL, LOG_DEBUG, "cleaning up completed thread");
-                        pthread_join(threads[i].tid, NULL);
-                        threads[i].tid = 0;
-                    }
-                }
+        /* Initialize server stuff */
+        if (server_init () < 0) {
+            ha_message (NULL, LOG_CRIT, "couldn't server internals");
+            exit (1);
+        }
 
-                /* Start a new thread if neccessary */
-                if(fd != -1 && threads[i].tid == 0)
-                {
-                    threads[i].fd = fd;
-                    r = pthread_create(&(threads[i].tid), NULL, httpauth_thread,
-                                       (void*)(threads + i));
-                    if(r != 0)
-                    {
-                        errno = r;
-                        ha_message(NULL, LOG_ERR, "couldn't create thread");
-                        g_quit = 1;
-                        break;
-                    }
-
-                    ha_messagex(NULL, LOG_DEBUG, "created thread for connection: %d", fd);
-                    fd = -1;
-                    break;
-                }
-            }
+        /* The thread pool */
+        if (tpool_init (g_maxthreads, g_minthreads, 1) < 0) {
+            ha_message (NULL, LOG_CRIT, "couldn't initialize thread pool");
+            exit (1);
+        }
 
-            /* Check to make sure we have a thread */
-            if(fd != -1)
-            {
-                ha_messagex(NULL, LOG_ERR, "too many connections open (max %d)", g_maxthreads);
-                httpauth_respond(NULL, fd, HA_SERVER_ERROR, 0, "too many connections");
-                shutdown(fd, SHUT_RDWR);
-            }
+        /* Wait on various messages */
+        if (server_watch (sock, SERVER_READ, accept_handler, NULL)) {
+            ha_message (NULL, LOG_CRIT, "couldn't watch socket properly");
+            exit (1);
         }
 
-        ha_messagex(NULL, LOG_INFO, "waiting for threads to quit");
+        if (server_run () < 0)
+            ha_message (NULL, LOG_ERR, "error running main loop");
 
-        /* Quit all threads here */
-        for(i = 0; i < g_maxthreads; i++)
-        {
-            /* Clean up quit threads */
-            if(threads[i].tid != 0)
-            {
-                if(threads[i].fd != -1)
-                {
-                    shutdown(threads[i].fd, SHUT_RDWR);
-                    close(threads[i].fd);
-                    threads[i].fd = -1;
-                }
+        close_all ();
 
-                pthread_join(threads[i].tid, NULL);
-            }
-        }
+        tpool_destroy (1);
+        server_uninit ();
 
         r = 0;
     }
@@ -477,7 +369,7 @@ finally:
 
 static void on_quit(int signal)
 {
-    g_quit = 1;
+    server_stop ();
     fprintf(stderr, "httpauthd: got signal to quit\n");
 }
 
@@ -505,703 +397,193 @@ static void writepid(const char* pidfile)
     }
 }
 
-static void* httpauth_thread(void* arg)
-{
-    httpauth_thread_t* thread = (httpauth_thread_t*)arg;
-    sigset_t set;
-    int r;
-
-    /* We handle these signals on the main thread */
-    sigemptyset(&set);
-    sigaddset(&set, SIGINT);
-    sigaddset(&set, SIGTERM);
-    pthread_sigmask(SIG_BLOCK, &set, NULL);
-
-    ASSERT(thread);
-    ASSERT(thread->fd != -1);
-
-    /* call the processor */
-    r = httpauth_processor(thread->fd, thread->fd);
-
-    /* mark this as done */
-    thread->fd = -1;
-    return (void*)r;
-}
-
 /* -----------------------------------------------------------------------
- * Logging
+ * Connection Handling
  */
 
-void log_request(ha_request_t* rq)
+static int
+write_data (int ofd, const char* data)
 {
-    const httpauth_command_t* cmd;
-    const char* t;
-    const char* t2;
-    int i;
+	int r;
 
-    if(g_debuglevel < LOG_DEBUG)
-        return;
+	assert (data);
+	assert (ofd != -1);
 
-    if(rq->req_type == REQTYPE_IGNORE || rq->req_type == -1)
-        return;
+	while (*data != 0) {
+		r = write (ofd, data, strlen (data));
+		if (r > 0)
+			data += r;
 
-    ha_bufcpy(rq->buf, "");
+		else if (r == -1) {
+			if(errno == EAGAIN)
+				continue;
 
-    for(i = 0; i < HA_MAX_ARGS; i++)
-    {
-        if(rq->req_args[i])
-        {
-            ha_bufjoin(rq->buf);
-            ha_bufmcat(rq->buf, ha_buflen(rq->buf) > 0 ? ", " : "", rq->req_args[i], NULL);
-        }
-    }
-
-    t = ha_bufdata(rq->buf);
-    t2 = NULL;
-
-    /* Figure out which command it is */
-    for(cmd = kCommands; cmd->name; cmd++)
-    {
-        if(cmd->code == rq->req_type)
-        {
-            t2 = cmd->name;
-            break;
-        }
-    }
-
-    ASSERT(t2);
-
-    ha_messagex(rq, LOG_DEBUG, "received request: [ type: %s / args: %s ]", t2, t);
+			/* The other end closed. no message */
+			if (errno != EPIPE)
+				ha_message(NULL, LOG_ERR, "couldn't write data");
+			return HA_CRITERROR;
+		}
+	}
 
-    for(i = 0; i < HA_MAX_HEADERS; i++)
-    {
-        if(rq->req_headers[i].name)
-        {
-            ASSERT(rq->req_headers[i].data);
-            ha_messagex(rq, LOG_DEBUG, "received header: [ %s: %s ]",
-                        rq->req_headers[i].name, rq->req_headers[i].data);
-        }
-    }
+	return HA_OK;
 }
 
-void log_response(ha_request_t* rq)
+/* Called when we cannot process a request */
+static void
+httpauth_respond_busy (int fd)
 {
-    int i;
+	char buf[16];
+	int l;
 
-    if(g_debuglevel < LOG_DEBUG)
-        return;
+	assert (fd != -1);
 
-    ha_messagex(rq, LOG_DEBUG, "sending response: [ code: 200 / ccode: %d / detail: %s ]",
-                rq->resp_code, rq->resp_detail ? rq->resp_detail : "");
+	/* Make it non blocking */
+	fcntl (fd, F_SETFL, fcntl (fd, F_GETFL, 0) | O_NONBLOCK);
 
-    for(i = 0; i < HA_MAX_HEADERS; i++)
-    {
-        if(rq->resp_headers[i].name)
-        {
-            ASSERT(rq->resp_headers[i].data);
-            ha_messagex(rq, LOG_DEBUG, "sending header: [ %s: %s ]",
-                        rq->resp_headers[i].name, rq->resp_headers[i].data);
-        }
-    }
-}
-
-void log_respcode(ha_request_t* rq, int code, const char* msg)
-{
-    if(g_debuglevel < LOG_DEBUG)
-        return;
+	/* Now read all the data sent from the client */
+	for(;;) {
+		l = read (fd, buf, sizeof (buf));
+		if (l <= 0)
+			break;
+	}
 
-    ha_messagex(rq, LOG_DEBUG, "sending response: [ code: %d / detail: %s ]",
-                code, msg ? msg : "");
+	/* Back to blocking mode */
+	fcntl (fd, F_SETFL, fcntl (fd, F_GETFL, 0) & ~O_NONBLOCK);
+	write_data (fd, "500 server too busy\n");
 }
 
-/* -----------------------------------------------------------------------
- * Command Parsing and Handling
+/*
+ * Called when a new connection is made, we initialize
+ * the new connection in a thread
  */
-
-static int httpauth_read(ha_request_t* rq, int ifd)
-{
-    const httpauth_command_t* cmd;
-    char* t;
-    int i, r;
-    int more = 1;
-
-    ASSERT(rq);
-    ASSERT(ifd != -1);
-
-    /* Clean up the request header */
-    rq->req_type = -1;
-    memset(rq->req_args, 0, sizeof(rq->req_args));
-    memset(rq->req_headers, 0, sizeof(rq->req_headers));
-
-
-    /* This guarantees a bit of memory allocated, and resets buffer */
-    ha_bufreset(rq->buf);
-
-    r = ha_bufreadline(ifd, rq->buf);
-    if(r == -1)
-    {
-        ha_message(rq, LOG_ERR, "error reading from socket");
-        return -1;
-    }
-
-    /* Check if this is the last line */
-    if(r == 0)
-        more = 0;
-
-    /* Check to see if we got anything */
-    if(ha_buflen(rq->buf) == 0)
-    {
-        rq->req_type = REQTYPE_IGNORE;
-        return more;
-    }
-
-    /* Find the first space in the line */
-    t = ha_bufparseword(rq->buf, " \t");
-
-    if(t)
-    {
-        /* Figure out which command it is */
-        for(cmd = kCommands; cmd->name; cmd++)
-        {
-            if(strcasecmp(t, cmd->name) == 0)
-            {
-                rq->req_type = cmd->code;
-                break;
-            }
-        }
-    }
-
-    else
-    {
-        rq->req_type = REQTYPE_IGNORE;
-        return more;
-    }
-
-    /* Check for invalid command */
-    if(rq->req_type == -1)
-        return more;
-
-    /* Now parse the arguments if any */
-    for(i = 0; i < cmd->word_args; i++)
-        rq->req_args[i] = ha_bufparseword(rq->buf, " \t");
-
-    /* Does it want the rest as one argument? */
-    if(cmd->rest_arg)
-        rq->req_args[i] = ha_bufparseline(rq->buf, 1);
-
-
-    /* Now skip anything else we have in the buffer */
-    ha_bufskip(rq->buf);
-
-
-    /* If we need headers, then read them now */
-    if(cmd->headers)
-    {
-        const char** head;  /* For iterating through valid headers */
-        int valid = 0;      /* The last header was valid */
-        i = 0;              /* The header we're working with */
-
-        for(;;)
-        {
-            /* Make sure we have more data */
-            if(!more)
-                break;
-
-            r = ha_bufreadline(ifd, rq->buf);
-            if(r == -1)
-            {
-                ha_message(rq, LOG_ERR, "error reading from socket");
-                return -1;
-            }
-
-            /* Check if this is the last line */
-            if(r == 0)
-                more = 0;
-
-            /* An empty line is the end of the headers */
-            if(ha_buflen(rq->buf) == 0)
-                break;
-
-            /* Check if the header starts with a space */
-            if(isspace(ha_bufchar(rq->buf)))
-            {
-                /* Skip all the spaces */
-                while(ha_buflen(rq->buf) > 0 && isspace(ha_bufchar(rq->buf)))
-                    ha_bufeat(rq->buf);
-
-                /* An empty line is the end of the headers
-                   even if that line has spaces on it */
-                if(ha_buflen(rq->buf) == 0)
-                    break;
-
-                /* A header that has data on it but started
-                   with a space continues the previous header */
-                if(valid && i > 0)
-                {
-                    t = ha_bufparseline(rq->buf, 0);
-                    if(t)
-                    {
-                        char* t2 = (char*)rq->req_headers[i - 1].data + strlen(rq->req_headers[i - 1].data);
-
-                        /* Fill the area between the end of the last
-                           valid header and this with spaces */
-                        memset(t2, ' ', t - t2);
-                    }
-                }
-            }
-            else
-            {
-                if(i < HA_MAX_HEADERS)
-                {
-                    t = ha_bufparseword(rq->buf, ":");
-
-                    if(t)
-                    {
-                        for(head = cmd->headers; ; head++)
-                        {
-                            if(!(*head))
-                            {
-                                t = NULL;
-                                break;
-                            }
-
-                            if(strcasecmp(t, *head) == 0)
-                                break;
-                        }
-                    }
-
-                    if(t)
-                    {
-                        rq->req_headers[i].data = ha_bufparseline(rq->buf, 1);
-
-                        /* We always need to have data for a header */
-                        if(rq->req_headers[i].data)
-                        {
-                            rq->req_headers[i].name = t;
-                            i++;
-                        }
-                    }
-
-                    valid = (t != NULL) ? 1 : 0;
-                }
-            }
-
-            ha_bufskip(rq->buf);
-        }
-    }
-
-    return more;
-}
-
-static int write_data(ha_request_t* rq, int ofd, const char* data)
-{
-    int r;
-
-    ASSERT(data);
-    ASSERT(ofd != -1);
-
-    while(*data != 0)
-    {
-        r = write(ofd, data, strlen(data));
-
-        if(r > 0)
-            data += r;
-
-        else if(r == -1)
-        {
-            if(errno == EAGAIN)
-                continue;
-
-            /* The other end closed. no message */
-            if(errno != EPIPE)
-                ha_message(rq, LOG_ERR, "couldn't write data");
-
-            return HA_CRITERROR;
-        }
-    }
-
-    return 0;
-}
-
-static int httpauth_respond(ha_request_t* rq, int ofd, int scode, int ccode, const char* msg)
+static void
+accept_handler (int sock, int type, void *arg)
 {
-    char num[16];
-
-    ASSERT(ofd != -1);
-    ASSERT(scode > 99 && scode < 1000);
-    ASSERT(ccode == 0 || (ccode > 99 && ccode < 1000));
-
-    /* Can only have a client code when server code is 200 */
-    ASSERT(ccode == 0 || scode == HA_SERVER_OK);
-
-    sprintf(num, "%d ", scode);
-
-    if(write_data(rq, ofd, num) < 0)
-        return HA_CRITERROR;
-
-    if(ccode != 0)
-    {
-        sprintf(num, "%d ", ccode);
-
-        if(write_data(rq, ofd, num) < 0)
-            return HA_CRITERROR;
-    }
-
-    if(!msg)
-    {
-        switch(scode)
-        {
-        case HA_SERVER_ACCEPTED:
-            msg = "Accepted";
-            break;
-        case HA_SERVER_ERROR:
-            msg = "Internal Error ";
-            break;
-        case HA_SERVER_BADREQ:
-            msg = "Bad Request ";
-            break;
-        case HA_SERVER_DECLINE:
-            msg = "Unauthorized ";
-            break;
-        default:
-            msg = NULL;
-            break;
-        };
-    }
-
-    if(msg && write_data(rq, ofd, msg) < 0)
-        return HA_CRITERROR;
-
-    /* When the client code is 0, then caller should log */
-    if(ccode == 0)
-        log_respcode(rq, scode, msg);
+	int fd;
+
+	fd = accept(sock, NULL, NULL);
+	if (fd == -1) {
+		switch (errno) {
+		case EINTR:
+		case EAGAIN:
+			return;
+
+		case ECONNABORTED:
+			ha_message (NULL, LOG_ERR, "couldn't accept a connection");
+			break;
+
+		default:
+			ha_message (NULL, LOG_CRIT, "couldn't accept a connection");
+			server_stop ();
+			break;
+		};
+	}
 
-    return write_data(rq, ofd, "\n");
+	/* Try to queue the request, or send back too busy if cannot */
+	if (tpool_add_work (ha_request_setup_handler, (void*)fd) < 0) {
+		ha_message (NULL, LOG_ERR, "too many requests active (max %d)", g_maxthreads);
+		httpauth_respond_busy (fd);
+		shutdown (fd, SHUT_RDWR);
+		close (fd);
+	}
 }
 
-const char kHeaderDelimiter[] = ": ";
-
-static int httpauth_write(ha_request_t* rq, int ofd)
+static void
+close_all (void)
 {
-    int i;
-    int wrote = 0;
-
-    ASSERT(ofd != -1);
-    ASSERT(rq);
-
-    if(httpauth_respond(rq, ofd, HA_SERVER_OK, rq->resp_code, rq->resp_detail) < 0)
-        return HA_CRITERROR;
-
-    for(i = 0; i < HA_MAX_HEADERS; i++)
-    {
-        if(rq->resp_headers[i].name)
-        {
-            if(write_data(rq, ofd, rq->resp_headers[i].name) == -1 ||
-               write_data(rq, ofd, kHeaderDelimiter) == -1 ||
-               write_data(rq, ofd, rq->resp_headers[i].data) == -1 ||
-               write_data(rq, ofd, "\n") == -1)
-                return -1;
-
-            wrote = 1;
-        }
-    }
+	hsh_index_t *hi;
+	ha_request_t *rq;
 
-    if(write_data(rq, ofd, "\n") == -1)
-        return -1;
+	ha_lock (NULL);
 
-    log_response(rq);
+	/* Get all the connections out of a wait state */
+	for (hi = hsh_first (g_requests); hi; hi = hsh_next (hi)) {
+		rq = hsh_this (hi, NULL);
+		shutdown (rq->ifd, SHUT_RDWR);
+	}
 
-    return 0;
+	ha_unlock (NULL);
 }
 
-static int httpauth_error(ha_request_t* rq, int ofd, int r)
+/*
+ * Called when a connection has data available on it, we
+ * process the data in a thread.
+ */
+static void
+httpauth_request_handler (int sock, int type, void *arg)
 {
-    int scode = 0;
-    const char* msg = NULL;
-
-    ASSERT(r < 0);
+	ha_request_t *rq = arg;
 
-    switch(r)
-    {
-    case HA_BADREQ:
-        scode = HA_SERVER_BADREQ;
-        break;
-
-    case HA_CRITERROR:
-        msg = "Critical Error";
-        /* fall through */
-
-    case HA_FAILED:
-        scode = HA_SERVER_ERROR;
-        break;
+	assert (arg);
 
-    default:
-        ASSERT(0 && "invalid error code");
-        break;
-    };
+	/* Unregister this socket, until ready for more */
+	server_unwatch (sock);
 
-    return httpauth_respond(rq, ofd, scode, 0, msg);
+	if (tpool_add_work (ha_request_process_handler, rq)) {
+		ha_message (NULL, LOG_ERR, "too many requests active (max %d)", g_maxthreads);
+		httpauth_respond_busy (rq->ofd);
+	}
 }
 
-static int httpauth_ready(ha_request_t* rq, int ofd)
+int
+ha_register_request (ha_request_t *rq)
 {
-    ASSERT(ofd != -1);
-    ASSERT(rq);
-
-    /* We send a ready banner to our client */
+	int ret = HA_OK;
 
-    if(CHECK_RBUF(rq))
-        return httpauth_error(rq, ofd, HA_CRITERROR);
+	ha_lock (NULL);
+		rq->id = ++g_unique;
+		if (!hsh_set (g_requests, &rq->id, rq)) {
+			ha_message (rq, LOG_ERR, "couldn't register new connection");
+			ret = HA_CRITERROR;
+		}
+	ha_unlock (NULL);
 
-    else
-        return httpauth_respond(rq, ofd, HA_SERVER_READY, 0, "HTTPAUTH/1.0");
+	return ret;
 }
 
-static int httpauth_auth(ha_request_t* rq, int ofd)
+int
+ha_register_watch (ha_request_t *rq)
 {
-    int r;
-
-    ASSERT(rq);
 
-    if(!rq->context)
-    {
-        ha_messagex(rq, LOG_ERR, "no auth handler set");
-        return httpauth_respond(rq, ofd, HA_SERVER_BADREQ, 0, "No Auth Handler Set");
-    }
+	int ret = HA_OK;
 
-    /* Clear out our response */
-    rq->resp_code = -1;
-    rq->resp_detail = NULL;
-    memset(rq->resp_headers, 0, sizeof(rq->resp_headers));
+	ha_lock (NULL);
+		if (server_watch (rq->ifd, SERVER_READ, httpauth_request_handler, rq) < 0) {
+			ha_message (rq, LOG_ERR, "couldn't watch new connection");
+			ret = HA_CRITERROR;
+		}
 
-    /* Check our connection argument */
-    if(!rq->req_args[AUTH_ARG_CONN] || !(rq->req_args[AUTH_ARG_CONN][0]))
-    {
-        ha_messagex(rq, LOG_ERR, "missing connection ID in request");
-        return httpauth_respond(rq, ofd, HA_SERVER_BADREQ, 0, "Missing Connection ID");
-    }
+	ha_unlock (NULL);
 
-    /* Check our uri argument */
-    if(!rq->req_args[AUTH_ARG_URI] || !(rq->req_args[AUTH_ARG_URI][0]))
-    {
-        ha_messagex(rq, LOG_ERR, "missing URI in request");
-        return httpauth_respond(rq, ofd, HA_SERVER_BADREQ, 0, "Missing URI");
-    }
-
-    /* Check our connection arguments */
-    if(!rq->req_args[AUTH_ARG_METHOD] || !(rq->req_args[AUTH_ARG_METHOD][0]))
-    {
-        ha_messagex(rq, LOG_ERR, "missing HTTP method in request");
-        return httpauth_respond(rq, ofd, HA_SERVER_BADREQ, 0, "Missing HTTP Method");
-    }
-
-    ASSERT(rq->context->handler && rq->context->handler->f_process);
-    r = (rq->context->handler->f_process)(rq);
-    if(r < 0)
-        return r;
-
-    if(httpauth_write(rq, ofd) < 0)
-        return HA_CRITERROR;
-
-    return HA_OK;
+	return ret;
 }
 
-static int httpauth_set(ha_request_t* rq, ha_buffer_t* cbuf, int ofd)
+void
+ha_unregister_request (ha_request_t *rq)
 {
-    httpauth_loaded_t* h;
-    const char* name = rq->req_args[0];
-    const char* value = rq->req_args[1];
-
-    /* Check our name argument */
-    if(!name || !*name)
-    {
-        ha_messagex(rq, LOG_ERR, "missing name in SET request");
-        return HA_BADREQ;
-    }
-
-    if(strcasecmp(name, "Domain") == 0)
-    {
-        /* We need to copy this string so it doesn't get destroyed on next req */
-        rq->digest_domain = ha_bufcpy(rq->conn_buf, value ? value : "");
-    }
-
-	else if (strcasecmp (name, "Groups") == 0) {
-
-		/* we need to copy this string so it doesn't get destroyed on next req */
-		if (rq->requested_groups)
-			str_array_free (rq->requested_groups);
-		rq->requested_groups = str_array_parse_quoted (value ? value : "");
-	}
-
-    else if(strcasecmp(name, "Handler") == 0)
-    {
-        if(!value || !*value)
-        {
-            ha_messagex(rq, LOG_ERR, "no auth handler specified in SET request.");
-            return HA_BADREQ;
-        }
-
-        /* Find a handler for this type */
-        for(h = g_handlers; h; h = h->next)
-        {
-            if(strcasecmp(h->ctx.name, value) == 0)
-            {
-                rq->context = &(h->ctx);
-                value = NULL;
-                break;
-            }
-        }
-
-        if(value != NULL)
-        {
-            ha_messagex(rq, LOG_ERR, "unknown authentication handler: %s", value);
-            return httpauth_respond(rq, ofd, HA_SERVER_BADREQ, 0, "Unknown Auth Handler");
-        }
-    }
-
-    else
-    {
-        ha_messagex(rq, LOG_ERR, "bad option in SET request");
-        return HA_BADREQ;
-    }
-
-    return httpauth_respond(rq, ofd, HA_SERVER_ACCEPTED, 0, NULL);
+	ha_lock (NULL);
+		hsh_rem (g_requests, &rq->id);
+	ha_unlock (NULL);
 }
 
-static void httpauth_conninfo(ha_request_t* rq, int fd)
+ha_context_t*
+ha_lookup_handler (const char *name)
 {
-    struct sockaddr_any addr;
-    char peername[MAXPATHLEN];
+	httpauth_loaded_t* h;
+	ha_context_t *ret = NULL;
 
-    ha_messagex(rq, LOG_DEBUG, "processing %d on thread %x", fd, (int)pthread_self());
+	assert (name);
 
-    memset(&addr, 0, sizeof(addr));
-    SANY_LEN(addr) = sizeof(addr);
+	ha_lock (NULL);
+		for (h = g_handlers; h; h = h->next) {
+			if (strcasecmp (h->ctx.name, name) == 0) {
+				ret = (&h->ctx);
+				break;
+			}
+		}
+	ha_unlock (NULL);
 
-    /* Get the peer name */
-    if(getpeername(fd, &SANY_ADDR(addr), &SANY_LEN(addr)) == -1 ||
-       sock_any_ntop(&addr, peername, MAXPATHLEN, SANY_OPT_NOPORT) == -1)
-        ha_messagex(rq, LOG_WARNING, "couldn't get peer address");
-    else
-        ha_messagex(rq, LOG_INFO, "accepted connection from: %s", peername);
-}
-
-static int httpauth_processor(int ifd, int ofd)
-{
-    ha_buffer_t cbuf;
-    ha_buffer_t buf;
-    ha_request_t rq;
-    int result = -1;
-    int r = 0;
-
-    ASSERT(ifd != -1);
-    ASSERT(ofd != -1);
-
-    memset(&rq, 0, sizeof(rq));
-
-    ha_lock(NULL);
-        rq.id = g_unique++;
-    ha_unlock(NULL);
-
-    /* Used when processing a socket */
-    if(ifd == ofd)
-        httpauth_conninfo(&rq, ifd);
-
-    /* Initialize the memory buffers */
-    ha_bufinit(&buf);
-    ha_bufinit(&cbuf);
-
-    /* Set up some context stuff */
-    rq.digest_domain = "";
-    rq.requested_groups = NULL;
-    rq.buf = &buf;
-    rq.conn_buf = &cbuf;
-
-    if(httpauth_ready(&rq, ofd) == -1)
-    {
-        result = 1;
-        goto finally;
-    }
-
-    /* Now loop and handle the commands */
-    while(result == -1)
-    {
-        ha_bufreset(&buf);
-
-        r = httpauth_read(&rq, ifd);
-
-        if(CHECK_RBUF(&rq))
-            r = HA_CRITERROR;
-
-        if(r < 0)
-        {
-            httpauth_error(&rq, ofd, r);
-            result = 1;
-            break;
-        }
-
-        log_request(&rq);
-
-        if(r == 0)
-            result = 0;
-
-        switch(rq.req_type)
-        {
-        case REQTYPE_AUTH:
-            r = httpauth_auth(&rq, ofd);
-            break;
-
-        case REQTYPE_SET:
-            r = httpauth_set(&rq, &cbuf, ofd);
-            break;
-
-        case REQTYPE_QUIT:
-            r = HA_OK;
-            result = 0;
-            break;
-
-        case REQTYPE_IGNORE:
-            r = HA_FALSE;
-            break;
-
-        default:
-            ha_messagex(&rq, LOG_WARNING, "received unknown command from client");
-            r = httpauth_respond(&rq, ofd, HA_SERVER_BADREQ, 0, "Unknown command");
-            break;
-        };
-
-        if(CHECK_RBUF(&rq))
-            r = HA_CRITERROR;
-
-        if(r < 0)
-        {
-            httpauth_error(&rq, ofd, r);
-
-            if(r == HA_CRITERROR)
-                result = 1;
-        }
-    }
-
-finally:
-
-    if(ifd == ofd)
-    {
-        shutdown(ifd, SHUT_RDWR);
-        close(ifd);
-    }
-    else
-    {
-        close(ifd);
-        close(ofd);
-    }
-
-    if (rq.requested_groups)
-	    str_array_free (rq.requested_groups);
-    ha_messagex(&rq, LOG_INFO, "closed connection");
-
-    ha_buffree(&cbuf);
-    ha_buffree(&buf);
-    return result;
+	return ret;
 }
 
 /* -----------------------------------------------------------------------
@@ -1410,7 +792,14 @@ static int config_parse(const char* file, ha_buffer_t* buf)
 
             else if(strcmp("maxthreads", name) == 0)
             {
-                if(ha_confint(name, value, 1, 1024, &g_maxthreads) == -1)
+                if(ha_confint(name, value, 0, 4096, &g_maxthreads) == -1)
+                    exit(1);
+                recog = 1;
+            }
+
+            else if(strcmp("minthreads", name) == 0)
+            {
+                if(ha_confint(name, value, 0, 4096, &g_minthreads) == -1)
                     exit(1);
                 recog = 1;
             }
diff --git a/daemon/httpauthd.h b/daemon/httpauthd.h
index e533dde..604c4db 100644
--- a/daemon/httpauthd.h
+++ b/daemon/httpauthd.h
@@ -22,6 +22,7 @@
 #ifndef __HTTPAUTHD_H__
 #define __HTTPAUTHD_H__
 
+#include <stdlib.h>
 #include "buffer.h"
 #include <syslog.h>
 
@@ -126,7 +127,7 @@ ha_context_t;
 
 
 /* -----------------------------------------------------------------------
- * HTTP Auth Structures and Constants
+ * Request Stuff
  */
 
 /*
@@ -175,6 +176,8 @@ ha_header_t;
 typedef struct ha_request
 {
     unsigned int id;                            /* Unique connection identifier */
+    int ifd;                                    /* Input file descriptor */
+    int ofd;                                    /* Output file descriptor */
 
     int req_type;                               /* The command type */
     const char* req_args[HA_MAX_ARGS];          /* Arguments for the command */
@@ -185,11 +188,14 @@ typedef struct ha_request
     const char* digest_domain;
     char** requested_groups;
 
+    /* Shortcut to req_buf below, for compatibility */
+    ha_buffer_t *buf;
+
     /* The buffer in use for the request */
-    ha_buffer_t* buf;
+    ha_buffer_t req_buf;
 
     /* The buffer in use for the connection */
-    ha_buffer_t* conn_buf;
+    ha_buffer_t conn_buf;
 
     int resp_code;                              /* The response code */
     const char* resp_detail;                    /* The details for response */
@@ -212,7 +218,25 @@ const char* ha_getheader(const ha_request_t* rq, const char* name, const char* p
 /* Response functions */
 void ha_addheader(ha_request_t* rq, const char* name, const char* data);
 
-/* Configuration functions */
+/* Implemented in request.c */
+void ha_request_destroy (ha_request_t *rq);
+ha_request_t* ha_request_setup (int ifd, int ofd);
+void ha_request_setup_handler (void *arg);
+int ha_request_process (ha_request_t *rq);
+void ha_request_process_handler (void *arg);
+void ha_request_loop (int ifd, int ofd);
+
+/* Implemented in httpauthd.c */
+int ha_register_request (ha_request_t *rq);
+int ha_register_watch (ha_request_t *rq);
+void ha_unregister_request (ha_request_t *rq);
+ha_context_t* ha_lookup_handler (const char *name);
+
+
+/* -----------------------------------------------------------------------
+ * Configuration
+ */
+
 int ha_confbool(const char* name, const char* conf, int* value);
 int ha_confint(const char* name, const char* conf, int min, int max, int* value);
 
@@ -261,13 +285,12 @@ char* ha_uriformat(ha_buffer_t* buf, const ha_uri_t* uri);
 int ha_uriparse(ha_buffer_t* buf, const char* suri, ha_uri_t* uri);
 int ha_uricmp(ha_uri_t* one, ha_uri_t* two);
 
-
 /* -----------------------------------------------------------------------
  * Locking
  */
 
-void ha_lock();
-void ha_unlock();
+void ha_lock (pthread_mutex_t* mtx);
+void ha_unlock (pthread_mutex_t* mtx);
 
 
 /* -----------------------------------------------------------------------
@@ -276,5 +299,4 @@ void ha_unlock();
 
 int ha_genrandom(unsigned char* data, size_t len);
 
-
 #endif /* __HTTPAUTHD_H__ */
diff --git a/daemon/request.c b/daemon/request.c
new file mode 100644
index 0000000..30c9d2f
--- /dev/null
+++ b/daemon/request.c
@@ -0,0 +1,752 @@
+/*
+ * HttpAuth
+ *
+ * Copyright (C) 2008 Stefan Walter
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the
+ * Free Software Foundation, Inc.,
+ * 59 Temple Place, Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#include "config.h"
+
+#include "httpauthd.h"
+#include "usuals.h"
+
+#include <sys/types.h>
+#include <sys/param.h>
+
+#include <assert.h>
+#include <ctype.h>
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "common/sock-any.h"
+#include "common/stringx.h"
+
+/* -----------------------------------------------------------------------
+ * Structures and Constants
+ */
+
+/* A command definition. Used in parsing */
+typedef struct httpauth_command
+{
+    const char* name;
+    int code;
+    int word_args;          /* Arguments to be parsed as words */
+    int rest_arg;           /* Parse remainder as one arg? */
+    const char** headers;   /* Headers needed/allowed */
+}
+httpauth_command_t;
+
+/* The various valid headers for the auth command */
+const char* kAuthHeaders[] =
+{
+    "Authorization",
+    NULL
+};
+
+/* The command definitions */
+const httpauth_command_t kCommands[] =
+{
+    { "auth", REQTYPE_AUTH, 3, 0, kAuthHeaders },
+    { "set",  REQTYPE_SET,  1, 1, NULL         },
+    { "quit", REQTYPE_QUIT, 0, 0, NULL         },
+    { NULL,   -1,           0, 0, NULL         }
+};
+
+static const char HEADER_DELIMITER[] = ": ";
+
+/*
+ * This shouldn't be used by handlers,
+ * they should return HA_FAILED instead.
+ */
+#define HA_SERVER_ERROR    500
+
+extern int g_debuglevel;
+
+/* ---------------------------------------------------------------------------
+ * Logging of Requests
+ */
+
+static void
+log_request (ha_request_t *rq)
+{
+	const httpauth_command_t *cmd;
+	const char *t;
+	const char *t2;
+	int i;
+
+	if (g_debuglevel < LOG_DEBUG)
+		return;
+
+	if (rq->req_type == REQTYPE_IGNORE || rq->req_type == -1)
+		return;
+
+	ha_bufcpy (rq->buf, "");
+
+	for (i = 0; i < HA_MAX_ARGS; i++) {
+		if (rq->req_args[i]) {
+			ha_bufjoin (rq->buf);
+			ha_bufmcat (rq->buf, ha_buflen (rq->buf) > 0 ? ", " : "", rq->req_args[i], NULL);
+		}
+	}
+
+	t = ha_bufdata (rq->buf);
+	t2 = NULL;
+
+	/* Figure out which command it is */
+	for (cmd = kCommands; cmd->name; cmd++) {
+		if (cmd->code == rq->req_type) {
+			t2 = cmd->name;
+			break;
+		}
+	}
+
+	assert (t2);
+	ha_messagex (rq, LOG_DEBUG, "received request: [ type: %s / args: %s ]", t2, t);
+
+	for (i = 0; i < HA_MAX_HEADERS; i++) {
+		if (rq->req_headers[i].name) {
+			assert (rq->req_headers[i].data);
+			ha_messagex (rq, LOG_DEBUG, "received header: [ %s: %s ]",
+			             rq->req_headers[i].name, rq->req_headers[i].data);
+		}
+	}
+}
+
+static void
+log_response (ha_request_t *rq)
+{
+	int i;
+
+	if (g_debuglevel < LOG_DEBUG)
+		return;
+
+	ha_messagex (rq, LOG_DEBUG, "sending response: [ code: 200 / ccode: %d / detail: %s ]",
+	             rq->resp_code, rq->resp_detail ? rq->resp_detail : "");
+
+	for (i = 0; i < HA_MAX_HEADERS; i++) {
+		if(rq->resp_headers[i].name) {
+			assert (rq->resp_headers[i].data);
+			ha_messagex (rq, LOG_DEBUG, "sending header: [ %s: %s ]",
+			             rq->resp_headers[i].name, rq->resp_headers[i].data);
+		}
+	}
+}
+
+void
+log_respcode (ha_request_t *rq, int code, const char* msg)
+{
+	if (g_debuglevel < LOG_DEBUG)
+		return;
+
+	ha_messagex (rq, LOG_DEBUG, "sending response: [ code: %d / detail: %s ]",
+	             code, msg ? msg : "");
+}
+
+/* ------------------------------------------------------------------------------
+ * Request Handling
+ */
+
+static int
+write_data (ha_request_t *rq, const char *data)
+{
+	int ret;
+
+	assert (rq);
+	assert (data);
+	assert (rq->ofd != -1);
+
+	while (*data != 0) {
+		ret = write (rq->ofd, data, strlen (data));
+
+		if (ret > 0)
+			data += ret;
+
+		else if (ret == -1) {
+			if (errno == EAGAIN)
+				continue;
+
+			/* The other end closed. no message */
+			if (errno != EPIPE)
+				ha_message (rq, LOG_ERR, "couldn't write data");
+
+			return HA_CRITERROR;
+		}
+	}
+
+	return HA_OK;
+}
+
+static int
+respond_code (ha_request_t *rq, int scode, int ccode, const char* msg)
+{
+	char num[16];
+
+	assert (rq->ofd != -1);
+	assert (scode > 99 && scode < 1000);
+	assert (ccode == 0 || (ccode > 99 && ccode < 1000));
+
+	/* Can only have a client code when server code is 200 */
+	assert (ccode == 0 || scode == HA_SERVER_OK);
+
+	sprintf (num, "%d ", scode);
+
+	if (write_data (rq, num) < 0)
+		return HA_CRITERROR;
+
+	if (ccode != 0) {
+		sprintf (num, "%d ", ccode);
+
+		if (write_data (rq, num) < 0)
+			return HA_CRITERROR;
+	}
+
+	if (!msg) {
+		switch (scode) {
+		case HA_SERVER_ACCEPTED:
+			msg = "Accepted";
+			break;
+		case HA_SERVER_ERROR:
+			msg = "Internal Error ";
+			break;
+		case HA_SERVER_BADREQ:
+			msg = "Bad Request ";
+			break;
+		case HA_SERVER_DECLINE:
+			msg = "Unauthorized ";
+			break;
+		default:
+			msg = NULL;
+			break;
+		};
+	}
+
+	if (msg && write_data (rq, msg) < 0)
+		return HA_CRITERROR;
+
+	/* When the client code is 0, then caller should log */
+	if (ccode == 0)
+		log_respcode (rq, scode, msg);
+
+	return write_data (rq, "\n");
+}
+
+static int
+respond_message (ha_request_t* rq)
+{
+	int i;
+	int wrote = 0;
+
+	assert (rq);
+	assert (rq->ofd != -1);
+
+	if (respond_code (rq, HA_SERVER_OK, rq->resp_code, rq->resp_detail) < 0)
+		return HA_CRITERROR;
+
+	for (i = 0; i < HA_MAX_HEADERS; i++) {
+		if(rq->resp_headers[i].name) {
+			if (write_data (rq, rq->resp_headers[i].name) == -1 ||
+			    write_data (rq, HEADER_DELIMITER) == -1 ||
+			    write_data (rq, rq->resp_headers[i].data) == -1 ||
+			    write_data (rq, "\n") == -1)
+				return HA_CRITERROR;
+		}
+
+		wrote = 1;
+	}
+
+	if (write_data (rq, "\n") == -1)
+		return HA_CRITERROR;
+
+	log_response (rq);
+
+	return HA_OK;
+}
+
+static int
+respond_error (ha_request_t *rq, int res)
+{
+	int scode = 0;
+	const char *msg = NULL;
+
+	assert (res < 0);
+
+	switch (res) {
+	case HA_BADREQ:
+		scode = HA_SERVER_BADREQ;
+		break;
+
+	case HA_CRITERROR:
+		msg = "Critical Error";
+		/* fall through */
+
+	case HA_FAILED:
+		scode = HA_SERVER_ERROR;
+		break;
+
+	default:
+		assert (0 && "invalid error code");
+		break;
+	};
+
+	return respond_code (rq, scode, 0, msg);
+}
+
+static int
+process_set (ha_request_t* rq)
+{
+	ha_context_t *ctx;
+	const char* name = rq->req_args[0];
+	const char* value = rq->req_args[1];
+
+	/* Check our name argument */
+	if (!name || !*name) {
+		ha_messagex (rq, LOG_ERR, "missing name in SET request");
+		return HA_BADREQ;
+	}
+
+	if (strcasecmp (name, "Domain") == 0) {
+		/* We need to copy this string so it doesn't get destroyed on next req */
+		rq->digest_domain = ha_bufcpy (&rq->conn_buf, value ? value : "");
+
+	} else if (strcasecmp (name, "Groups") == 0) {
+
+		/* we need to copy this string so it doesn't get destroyed on next req */
+		if (rq->requested_groups)
+			str_array_free (rq->requested_groups);
+		rq->requested_groups = str_array_parse_quoted (value ? value : "");
+
+	} else if (strcasecmp(name, "Handler") == 0) {
+
+		if (!value || !*value) {
+			ha_messagex (rq, LOG_ERR, "no auth handler specified in SET request.");
+			return HA_BADREQ;
+		}
+
+		ctx = ha_lookup_handler (value);
+		if (ctx == NULL) {
+			ha_messagex (rq, LOG_ERR, "unknown authentication handler: %s", value);
+			return respond_code (rq, HA_SERVER_BADREQ, 0, "Unknown Auth Handler");
+		}
+
+		rq->context = ctx;
+
+	} else {
+		ha_messagex (rq, LOG_ERR, "bad option in SET request");
+		return HA_BADREQ;
+	}
+
+	return respond_code (rq, HA_SERVER_ACCEPTED, 0, NULL);
+}
+
+static int
+process_auth (ha_request_t* rq)
+{
+	int ret;
+
+	assert (rq);
+
+	if (!rq->context) {
+		ha_messagex (rq, LOG_ERR, "no auth handler set");
+		return respond_code (rq, HA_SERVER_BADREQ, 0, "No Auth Handler Set");
+	}
+
+	/* Clear out our response */
+	rq->resp_code = -1;
+	rq->resp_detail = NULL;
+	memset (rq->resp_headers, 0, sizeof(rq->resp_headers));
+
+	/* Check our connection argument */
+	if (!rq->req_args[AUTH_ARG_CONN] || !(rq->req_args[AUTH_ARG_CONN][0])) {
+		ha_messagex (rq, LOG_ERR, "missing connection ID in request");
+		return respond_code (rq, HA_SERVER_BADREQ, 0, "Missing Connection ID");
+	}
+
+	/* Check our uri argument */
+	if (!rq->req_args[AUTH_ARG_URI] || !(rq->req_args[AUTH_ARG_URI][0])) {
+		ha_messagex (rq, LOG_ERR, "missing URI in request");
+		return respond_code (rq, HA_SERVER_BADREQ, 0, "Missing URI");
+	}
+
+	/* Check our method arguments */
+	if (!rq->req_args[AUTH_ARG_METHOD] || !(rq->req_args[AUTH_ARG_METHOD][0])) {
+		ha_messagex (rq, LOG_ERR, "missing HTTP method in request");
+		return respond_code (rq, HA_SERVER_BADREQ, 0, "Missing HTTP Method");
+	}
+
+	assert (rq->context->handler && rq->context->handler->f_process);
+	ret = (rq->context->handler->f_process)(rq);
+	if (ret < 0)
+		return ret;
+
+	return respond_message (rq);
+}
+
+static int
+read_request (ha_request_t* rq)
+{
+	const httpauth_command_t *cmd;
+	char *t;
+	int i, r;
+	int more = 1;
+
+	assert (rq);
+	assert (rq->ifd != -1);
+
+	/* Clean up the request header */
+	rq->req_type = -1;
+	memset (rq->req_args, 0, sizeof (rq->req_args));
+	memset (rq->req_headers, 0, sizeof (rq->req_headers));
+
+
+	/* This guarantees a bit of memory allocated, and resets buffer */
+	ha_bufreset (rq->buf);
+
+	r = ha_bufreadline (rq->ifd, rq->buf);
+	if (r == -1) {
+		ha_message(rq, LOG_ERR, "error reading from socket");
+		return -1;
+	}
+
+	/* Check if this is the last line */
+	if (r == 0)
+		more = 0;
+
+	/* Check to see if we got anything */
+	if (ha_buflen (rq->buf) == 0) {
+		rq->req_type = REQTYPE_IGNORE;
+		return more;
+	}
+
+	/* Find the first space in the line */
+	t = ha_bufparseword (rq->buf, " \t");
+	if (t) {
+		/* Figure out which command it is */
+		for (cmd = kCommands; cmd->name; cmd++) {
+			if (strcasecmp (t, cmd->name) == 0) {
+				rq->req_type = cmd->code;
+				break;
+			}
+		}
+	} else {
+		rq->req_type = REQTYPE_IGNORE;
+		return more;
+	}
+
+	/* Check for invalid command */
+	if (rq->req_type == -1)
+		return more;
+
+	/* Now parse the arguments if any */
+	for (i = 0; i < cmd->word_args; i++)
+		rq->req_args[i] = ha_bufparseword (rq->buf, " \t");
+
+	/* Does it want the rest as one argument? */
+	if (cmd->rest_arg)
+		rq->req_args[i] = ha_bufparseline (rq->buf, 1);
+
+	/* Now skip anything else we have in the buffer */
+	ha_bufskip (rq->buf);
+
+	/* If we need headers, then read them now */
+	if (cmd->headers) {
+		const char **head;  /* For iterating through valid headers */
+		int valid = 0;      /* The last header was valid */
+		i = 0;              /* The header we're working with */
+
+		for (;;) {
+			/* Make sure we have more data */
+			if (!more)
+				break;
+
+			r = ha_bufreadline (rq->ifd, rq->buf);
+			if (r == -1) {
+				ha_message (rq, LOG_ERR, "error reading from socket");
+				return -1;
+			}
+
+			/* Check if this is the last line */
+			if (r == 0)
+				more = 0;
+
+			/* An empty line is the end of the headers */
+			if (ha_buflen (rq->buf) == 0)
+				break;
+
+			/* Check if the header starts with a space */
+			if (isspace (ha_bufchar (rq->buf))) {
+				/* Skip all the spaces */
+				while (ha_buflen (rq->buf) > 0 && isspace (ha_bufchar (rq->buf)))
+					ha_bufeat (rq->buf);
+
+				/*
+				 * An empty line is the end of the headers
+				 * even if that line has spaces on it
+				 */
+				if (ha_buflen (rq->buf) == 0)
+					break;
+
+				/*
+				 * A header that has data on it but started
+				 * with a space continues the previous header
+				 */
+				if (valid && i > 0) {
+					t = ha_bufparseline (rq->buf, 0);
+					if(t) {
+						char* t2 = (char*)rq->req_headers[i - 1].data + strlen (rq->req_headers[i - 1].data);
+
+						/*
+						 * Fill the area between the end of the last
+						 * valid header and this with spaces
+						 */
+						memset (t2, ' ', t - t2);
+					}
+				}
+			} else {
+				if (i < HA_MAX_HEADERS) {
+					t = ha_bufparseword (rq->buf, ":");
+
+					if(t) {
+						for (head = cmd->headers; ; head++) {
+							if (!(*head)) {
+								t = NULL;
+								break;
+							}
+							if (strcasecmp(t, *head) == 0)
+								break;
+						}
+					}
+
+					if (t) {
+						rq->req_headers[i].data = ha_bufparseline (rq->buf, 1);
+
+						/* We always need to have data for a header */
+						if (rq->req_headers[i].data) {
+							rq->req_headers[i].name = t;
+							i++;
+						}
+					}
+
+					valid = (t != NULL) ? 1 : 0;
+				}
+			}
+
+			ha_bufskip (rq->buf);
+		}
+	}
+
+	return more;
+}
+
+static void
+log_conninfo (ha_request_t* rq)
+{
+	struct sockaddr_any addr;
+	char peername[MAXPATHLEN];
+
+	memset (&addr, 0, sizeof (addr));
+	SANY_LEN (addr) = sizeof (addr);
+
+	/* Get the peer name */
+	if (getpeername (rq->ifd, &SANY_ADDR (addr), &SANY_LEN (addr)) == -1 ||
+	    sock_any_ntop (&addr, peername, MAXPATHLEN, SANY_OPT_NOPORT) == -1)
+		ha_messagex (rq, LOG_WARNING, "couldn't get peer address");
+	else
+		ha_messagex (rq, LOG_INFO, "accepted connection from: %s", peername);
+}
+
+void
+ha_request_destroy (ha_request_t *rq)
+{
+	assert (rq);
+
+	ha_unregister_request (rq);
+
+	/* A socket style connection */
+	if (rq->ifd == rq->ofd) {
+		if (rq->ifd != -1) {
+			shutdown (rq->ifd, SHUT_RDWR);
+			close (rq->ifd);
+			rq->ifd = rq->ofd = -1;
+		}
+
+	/* A pipe style connection */
+	} else {
+		if (rq->ifd != -1)
+			close (rq->ifd);
+		if (rq->ofd != -1)
+			close (rq->ofd);
+		rq->ifd = rq->ofd = -1;
+	}
+
+	if (rq->requested_groups)
+		str_array_free (rq->requested_groups);
+
+	ha_messagex (rq, LOG_INFO, "closed connection");
+
+	ha_buffree (&rq->conn_buf);
+	ha_buffree (&rq->req_buf);
+	rq->buf = NULL;
+
+	free (rq);
+}
+
+ha_request_t*
+ha_request_setup (int ifd, int ofd)
+{
+	int ret;
+	ha_request_t *rq;
+
+	assert (ifd != -1);
+	assert (ofd != -1);
+
+	rq = calloc (1, sizeof (ha_request_t));
+	if (rq == NULL) {
+		ha_memerr (NULL);
+		return NULL;
+	}
+
+       	rq->ifd = ifd;
+       	rq->ofd = ofd;
+
+       	/* Initialize the memory buffers */
+       	ha_bufinit (&rq->req_buf);
+       	ha_bufinit (&rq->conn_buf);
+       	rq->buf = &rq->req_buf;
+
+	/* Unique identifier for the request */
+       	if (ha_register_request (rq) < 0) {
+       		ha_request_destroy (rq);
+       		return NULL;
+       	}
+
+       	/* Used when processing a socket */
+       	if (ifd == ofd)
+       		log_conninfo (rq);
+
+       	/* Set up some context stuff */
+       	rq->digest_domain = "";
+       	rq->requested_groups = NULL;
+
+       	/* We send a ready banner to our client */
+       	if (CHECK_RBUF (rq))
+       		ret = respond_error (rq, HA_CRITERROR);
+       	else
+       		ret = respond_code (rq, HA_SERVER_READY, 0, "HTTPAUTH/1.0");
+
+       	if (ret < 0) {
+       		ha_message (rq, LOG_ERR, "couldn't handshake new connection");
+       		ha_request_destroy (rq);
+       		rq = NULL;
+       	}
+
+       	return rq;
+}
+
+void
+ha_request_setup_handler (void *arg)
+{
+	ha_request_t *rq = arg;
+	int fd = (int)arg;
+
+	/* This closes the connections on failure */
+	rq = ha_request_setup (fd, fd);
+	if (rq)
+		ha_register_watch (rq);
+}
+
+int
+ha_request_process (ha_request_t *rq)
+{
+	int ret, cont = 1;
+
+	ha_bufreset (&rq->req_buf);
+
+	ret = read_request (rq);
+	if (CHECK_RBUF (rq))
+        	ret = HA_CRITERROR;
+
+	if (ret < 0) {
+		respond_error (rq, ret);
+		return ret != HA_CRITERROR;
+	}
+
+	log_request (rq);
+
+	switch(rq->req_type) {
+	case REQTYPE_AUTH:
+		ret = process_auth (rq);
+		break;
+
+	case REQTYPE_SET:
+		ret = process_set (rq);
+		break;
+
+	case REQTYPE_QUIT:
+		ret = HA_OK;
+		cont = 0;
+		break;
+
+	case REQTYPE_IGNORE:
+		ret = HA_FALSE;
+		break;
+
+	default:
+		ha_messagex (rq, LOG_WARNING, "received unknown command from client");
+		ret = respond_code (rq, HA_SERVER_BADREQ, 0, "Unknown command");
+		break;
+	};
+
+	if (CHECK_RBUF (rq))
+		ret = HA_CRITERROR;
+
+	if (ret < 0) {
+		respond_error (rq, ret);
+		if (ret == HA_CRITERROR)
+			cont = 0;
+	}
+
+	return cont;
+}
+
+void
+ha_request_process_handler (void *arg)
+{
+	ha_request_t *rq = arg;
+
+	if (ha_request_process (rq))
+		ha_register_watch (rq);
+	else
+		ha_request_destroy (rq);
+}
+
+void
+ha_request_loop (int ifd, int ofd)
+{
+	ha_request_t *rq;
+	int cont = 1;
+
+	rq = ha_request_setup (ifd, ofd);
+	if (!rq)
+		return;
+
+	while (cont)
+		cont = ha_request_process (rq);
+
+	ha_request_destroy (rq);
+}