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

   connection, but instead only use threads for active requests.

1 files changed, 183 insertions, 794 deletions

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