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

   connection, but instead only use threads for active requests.

4 files changed, 967 insertions, 803 deletions

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);
+}