[PATCH] Naive approach to HTTP-timeouts

Index: gwlib/http.c
===================================================================
RCS file: /home/cvs/gateway/gwlib/http.c,v
retrieving revision 1.211
diff -u -p -b -r1.211 http.c
--- gwlib/http.c	15 Nov 2003 13:14:23 -0000	1.211
+++ gwlib/http.c	24 Nov 2003 10:34:18 -0000
@@ -95,11 +95,13 @@
 
 
 /*
- * Default port to connect to for HTTP connections.
+* Default port to connect to for HTTP connections.
  */
 enum { HTTP_PORT = 80,
        HTTPS_PORT = 443 };
 
+static Dict *active_clients;
+static Mutex *active_clients_lock;
 
 /*
  * Status of this module.
@@ -174,7 +176,6 @@ static int parse_http_version(Octstr *ve
     return 1;
 }
 
-
 /***********************************************************************
  * Proxy support.
  */
@@ -1693,6 +1694,15 @@ static void client_shutdown(void)
     fdset_destroy(client_fdset);
 }
 
+Octstr *pointer_to_octstr(void *p) {
+    Octstr *tmp_oct;
+    char *tmp = (char *) gw_malloc(sizeof(char) * 255);
+    sprintf(tmp, "%ld", p);
+    tmp_oct = octstr_create(tmp);
+    gw_free(tmp);
+    return tmp_oct;
+};
+
 
 /***********************************************************************
  * HTTP server interface.
@@ -1710,7 +1720,8 @@ struct HTTPClient {
         reading_request_line,
         reading_request,
         request_is_being_handled,
-        sending_reply
+        sending_reply,
+        to_be_destroyed
     } state;
     int method;  /* HTTP_METHOD_ value */
     Octstr *url;
@@ -1718,11 +1729,17 @@ struct HTTPClient {
     int persistent_conn;
     unsigned long conn_time; /* store time for timeouting */
     HTTPEntity *request;
+    Counter *ref_counter;
 };
 
-
+/*
+ * creates a new HTTPClient.
+ * Furthermore the client is registered in the active-clients dictionary.
+ */
 static HTTPClient *client_create(int port, Connection *conn, Octstr *ip)
 {
+    Octstr *key;
+    
     HTTPClient *p;
     
 #ifdef HAVE_LIBSSL
@@ -1743,18 +1760,29 @@ static HTTPClient *client_create(int por
     p->persistent_conn = 1;
     p->conn_time = time(NULL);
     p->request = NULL;
+    p->ref_counter = counter_create();
+
+    key = pointer_to_octstr(p);
+    dict_put(active_clients, key, p);
+    octstr_destroy(key);
     return p;
 }
 
-
-static void client_destroy(void *client)
-{
+/*
+ * frees resources allocated for a client.
+ */
+static void client_no_lock_destroy(void *client) {
     HTTPClient *p;
     
     if (client == NULL)
     	return;
-
     p = client;
+
+    if (counter_value(p->ref_counter) > 0) {
+        debug("", 0, "Cannot kill client! Client %p is still referenced.", client);
+        return;
+    };
+
     debug("gwlib.http", 0, "HTTP: Destroying HTTPClient area %p.", p);
     gw_assert_allocated(p, __FILE__, __LINE__, __func__);
     debug("gwlib.http", 0, "HTTP: Destroying HTTPClient for `%s'.",
@@ -1763,17 +1791,41 @@ static void client_destroy(void *client)
     octstr_destroy(p->ip);
     octstr_destroy(p->url);
     entity_destroy(p->request);
+    counter_destroy(p->ref_counter);
     gw_free(p);
-}
+    p = NULL;
+};
 
+/*
+ * destroys the client and removes it from the active-clients-dictionary.
+ */
+static void client_destroy(void *client) {
+    Octstr *key;
+    mutex_lock(active_clients_lock);
+    key = pointer_to_octstr(client);
+    if (dict_remove(active_clients, key) == NULL) {
+        octstr_destroy(key);
+        mutex_unlock(active_clients_lock);
+        return;
+    };
+    octstr_destroy(key);
+    client_no_lock_destroy(client);
+    mutex_unlock(active_clients_lock);
+};
 
 static void client_reset(HTTPClient *p)
 {
+    Octstr *key;
+    key = pointer_to_octstr(p);
+    
     debug("gwlib.http", 0, "HTTP: Resetting HTTPClient for `%s'.",
     	  octstr_get_cstr(p->ip));
     p->state = reading_request_line;
     p->conn_time = time(NULL);
     gw_assert(p->request == NULL);
+    counter_set(p->ref_counter, 0L);
+    dict_put(active_clients, key, p);
+    octstr_destroy(key);
 }
 
 
@@ -1982,24 +2034,36 @@ error:
     return -1;
 }
 
-
+/*
+ * Callback routine for http-requests.
+ * Implements a state machine:
+ * reading_request_line -> reading_request 
+ * -> request_is_being_handled -> sending_reply
+ *
+ * To avoid the destruction of active clients waiting for a reply,
+ * those clients are removed from the active-clients dictionary
+ * upon a reading_request -> request_is_being_handled transition.
+ */
 static void receive_request(Connection *conn, void *data)
 {
     HTTPClient *client;
     Octstr *line;
+    Octstr *key;
     int ret;
 
+    client = data;
     if (run_status != running) {
 	conn_unregister(conn);
 	return;
     }
-
-    client = data;
-    
     for (;;) {
+        debug("",0,"loop start: %p", client);
+            
 	switch (client->state) {
 	case reading_request_line:
     	    line = conn_read_line(conn);
+                client->conn_time = time(NULL);
+                /*debug("", 0, "reading_request_line");*/
 	    if (line == NULL) {
 		if (conn_eof(conn) || conn_read_error(conn))
 		    goto error;
@@ -2008,8 +2072,10 @@ static void receive_request(Connection *
 	    ret = parse_request_line(&client->method, &client->url,
                                      &client->use_version_1_0, line);
 	    octstr_destroy(line);
-	    if (ret == -1)
+                if (ret == -1) {
 	    	goto error;
+                }
+                
    	    /*
 	     * RFC2616 (4.3) says we should read a message body if there
 	     * is one, even on GET requests.
@@ -2019,10 +2085,15 @@ static void receive_request(Connection *
 	    break;
 	    
 	case reading_request:
+                /*debug("", 0, "reading_request");*/
 	    ret = entity_read(client->request, conn);
-	    if (ret < 0)
+                client->conn_time = time(NULL);
+                if (ret < 0) {
 		goto error;
+                }
 	    if (ret == 0) {
+                    /*dict_remove(active_clients, pointer_to_octstr(client));*/
+                    counter_increase(client->ref_counter);
 	    	client->state = request_is_being_handled;
 		conn_unregister(conn);
 		port_put_request(client);
@@ -2030,10 +2101,13 @@ static void receive_request(Connection *
 	    return;
 
 	case sending_reply:
-	    if (conn_outbuf_len(conn) > 0)
+                /*debug("", 0, "sending_reply");*/
+                if (conn_outbuf_len(conn) > 0) {
 		return;
+                }
 	    /* Reply has been sent completely */
 	    if (!client->persistent_conn) {
+                    counter_decrease(client->ref_counter);
                 conn_unregister(conn);
 		client_destroy(client);
 		return;
@@ -2043,8 +2117,9 @@ static void receive_request(Connection *
 	    break;
 
     	default:
-	    panic(0, "Internal error: HTTPClient state is wrong.");
+                return;
 	}
+        debug("",0,"loop end");
     }
     
 error:
@@ -2058,12 +2133,112 @@ struct server {
     int ssl;
 };
 
+/* check_timeout. Checks for timedout connections.
+ * each active clients (determined by using the active_clients-dictionary)
+ * is handled based on its state. Depending on the state different timeout-
+ * thresholds are used.
+ *
+ * NOTE: Some state transitions happen elsewhere in this module. 
+ * 
+ * state                    | timeout used          | action
+ * """"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
+ * reading_request_line     | CONNECTION_TIME_OUT   | - the client has not submitted a
+ *                                                    request within the timeout-period.
+ *                                                    this is used for first-time 
+ *                                                    communication only.
+ *                                                    - unregister the client.
+ *                                                    STATE = to_be_destroyed.
+ * reading_request          | IDLE_TIME_OUT         | - the communication has started, but 
+ *                                                    there has not been any input within the
+ *                                                    timeout period.
+ *                                                    - unregister the client.
+ *                                                    STATE = to_be_destroyed.
+ * request_is_being_handled | none                  | - A valid request has been submitted,
+ *                                                    thus the client will not be interrupted.
+ *                                                    STATE = request_is_being_handled.
+ * sending_reply            | none                  | - An answer is being sent to the client,
+ *                                                    thus the client will not be interrupted.
+ *                                                    STATE = sending_reply.
+ * to_be_destroyed          | none                  | - The client has already been unregistered and
+ *                                                    is now queued for destruction.
+ *                                                    - the client is removed from the list of active
+ *                                                    clients and subsequently destroyed.
+ */
+static void check_timeout(void) 
+{
+    List *keys;
+    int i;
+    HTTPClient *tmp_client;
+    time_t now;
+    double timediff;
+
+    debug("", 0, "Checking timeout...");
+    /*
+     * for each active client check status.
+     * - based on status check either CONNECTION or IDLE timeout.
+     * - if the client is timedout, then unregister the callback and
+     * - set the client's state to "to_be_destroyed"
+     */
+    mutex_lock(active_clients_lock);
+    now = time(NULL);
+    keys = dict_keys(active_clients);
+    for (i = 0; i < list_len(keys); i++) {
+        tmp_client = dict_get(active_clients, list_get(keys, i));
+        /*debug("", 0, "sleep now %p", tmp_client);*/
+        if (tmp_client == NULL)
+            continue;
+        /*gwthread_sleep(10);
+        debug("", 0, "awake now %p", tmp_client);*/
+        
+        debug("", 0, "Client:   %p", tmp_client);
+        debug("", 0, "Num ref:  %ld", counter_value(tmp_client->ref_counter));
+
+        if (counter_value(tmp_client->ref_counter) > 0)
+            continue;
+        timediff = difftime(now, tmp_client->conn_time);
+
+        switch (tmp_client->state) {
+            case reading_request_line:
+                debug("", 0, "State:   reading_request_line");
+                if (timediff > CONNECTION_TIME_OUT) {
+                    debug("", 0, "CONNECT TIMEOUT");
+                    tmp_client->state = to_be_destroyed;
+                    conn_unregister(tmp_client->conn);
+                }
+                break;
+            case reading_request:
+                debug("", 0, "State:   reading_request");
+                if (timediff > IDLE_TIME_OUT) {
+                    debug("", 0, "IDLE TIMEOUT");
+                    tmp_client->state = to_be_destroyed;
+                    conn_unregister(tmp_client->conn);
+                }
+                break;
+            case request_is_being_handled:
+                debug("", 0, "State:   request_is_being_handled [ignored]");
+                break;
+            case sending_reply:
+                debug("", 0, "State:   sending_reply [ignored]");
+                break;
+            case to_be_destroyed:
+                debug("", 0, "State:   to_be_destroyed");
+                dict_remove(active_clients, list_get(keys, i));
+                client_no_lock_destroy(tmp_client);
+                break;
+            default:
+                panic(0, "Invalid state %d", tmp_client->state);
+        };
+    }
+    mutex_unlock(active_clients_lock);
+    debug("", 0, "timeout check done!");
+};
 
 static void server_thread(void *dummy)
 {
     struct pollfd tab[MAX_SERVERS];
     int ports[MAX_SERVERS];
     int ssl[MAX_SERVERS];
+
     long i, j, n, fd;
     int *portno;
     struct server *p;
@@ -2090,10 +2265,21 @@ static void server_thread(void *dummy)
             gw_free(p);
         }
 
-        if ((ret = gwthread_poll(tab, n, -1.0)) == -1) {
+        /* poll returns -1 on error and 0 on timeout. we'll handle both */
+        if ((ret = gwthread_poll(tab, n, MIN_TIME_OUT)) < 1) {
+            switch(ret) {
+            case 0: /* timeout */
+                    /*debug("", 0, "check tout");*/
+                    check_timeout();
+                break;
+            case -1: /* error */
             if (errno != EINTR) /* a signal was caught during poll() function */
                 warning(0, "HTTP: gwthread_poll failed.");
             continue;
+                break;
+            default: /* should not happen */
+                panic(0, "This is not supposed to happen!");
+            }
         }
 
         for (i = 0; i < n; ++i) {
@@ -2178,7 +2364,6 @@ static void start_server_thread(void)
     }
 }
 
-
 int http_open_port_if(int port, int ssl, Octstr *interface)
 {
     struct server *p;
@@ -2288,8 +2473,8 @@ HTTPClient *http_accept_request(int port
 				List **cgivars)
 {
     HTTPClient *client;
-
     client = port_get_request(port);
+
     if (client == NULL) {
 	debug("gwlib.http", 0, "HTTP: No clients with requests, quitting.");
     	return NULL;
@@ -2314,7 +2499,6 @@ HTTPClient *http_accept_request(int port
     client->request->body = NULL;
     entity_destroy(client->request);
     client->request = NULL;
-
     return client;
 }
 
@@ -2359,6 +2543,7 @@ void http_send_reply(HTTPClient *client,
     if (ret == 0) { 
         /* HTTP/1.0 or 1.1, hence keep-alive or keep-alive */
         if (!client->persistent_conn) {
+            counter_decrease(client->ref_counter);
             client_destroy(client);     
         } else {
             /* XXX mark this HTTPClient in the keep-alive cleaner thread */
@@ -2373,6 +2558,7 @@ void http_send_reply(HTTPClient *client,
     }
     /* error while sending response */
     else {     
+        counter_decrease(client->ref_counter);
         client_destroy(client);
     }
 }
@@ -3179,6 +3365,9 @@ void http_init(void)
 {
     gw_assert(run_status == limbo);
 
+    active_clients = dict_create(1024, NULL);
+    active_clients_lock = mutex_create();
+
 #ifdef HAVE_LIBSSL
     openssl_init_locks();
     conn_init_ssl();
@@ -3195,6 +3384,23 @@ void http_init(void)
     run_status = running;
 }
 
+void purge_active_clients() 
+{
+    List *keys;
+    int i;
+    HTTPClient *tmp_client;
+
+    debug("", 0, "Purging clients...");
+    keys = dict_keys(active_clients);
+    for (i = 0; i < list_len(keys); ++i) {
+        tmp_client = dict_get(active_clients, list_get(keys, i));       
+        if (tmp_client != NULL) {
+            conn_flush_real(tmp_client->conn, 2.0);
+            conn_unregister(tmp_client->conn);
+        }
+        client_destroy(tmp_client);
+    }   
+};
 
 void http_shutdown(void)
 {
@@ -3203,10 +3409,13 @@ void http_shutdown(void)
 
     run_status = terminating;
 
+    mutex_destroy(active_clients_lock);
     conn_pool_shutdown();
     port_shutdown();
     client_shutdown();
     server_shutdown();
+    purge_active_clients();
+    dict_destroy(active_clients);
     proxy_shutdown();
 #ifdef HAVE_LIBSSL
     openssl_shutdown_locks();
Index: gwlib/http.h
===================================================================
RCS file: /home/cvs/gateway/gwlib/http.h,v
retrieving revision 1.60
diff -u -p -b -r1.60 http.h
--- gwlib/http.h	15 Nov 2003 13:14:23 -0000	1.60
+++ gwlib/http.h	24 Nov 2003 10:34:18 -0000
@@ -194,13 +194,21 @@ typedef struct {
 	Octstr *value;
 } HTTPCGIVar;
 
+/*
+ * Connection resp. flush timeout in seconds. 
+ */
+enum {
+    MIN_TIME_OUT=10,
+    CONNECTION_TIME_OUT=50,
+    IDLE_TIME_OUT=10,
+    FLUSH_TIME_OUT=2
+};
 
 /*
  * Initialization function. This MUST be called before any other function
  * declared in this header file.
  */
 void http_init(void);
-
 
 /*
  * Shutdown function. This MUST be called when no other function
Index: gwlib/conn.c
===================================================================
RCS file: /home/cvs/gateway/gwlib/conn.c,v
retrieving revision 1.67
diff -u -p -b -r1.67 conn.c
--- gwlib/conn.c	15 Nov 2003 13:14:23 -0000	1.67
+++ gwlib/conn.c	24 Nov 2003 10:34:18 -0000
@@ -1058,7 +1058,8 @@ int conn_wait(Connection *conn, double s
     return 0;
 }
 
-int conn_flush(Connection *conn)
+
+int conn_flush_real(Connection *conn, double tout_seconds)
 {
     int ret;
     int revents;
@@ -1075,7 +1076,7 @@ int conn_flush(Connection *conn)
         fd = conn->fd;
 
         unlock_out(conn);
-        revents = gwthread_pollfd(fd, POLLOUT, -1.0);
+        revents = gwthread_pollfd(fd, POLLOUT, tout_seconds);
 
         /* Note: Make sure we have the "out" lock when
          * going through the loop again, because the 
@@ -1113,6 +1114,7 @@ int conn_flush(Connection *conn)
 
     return 0;
 }
+
 
 int conn_write(Connection *conn, Octstr *data)
 {
Index: gwlib/conn.h
===================================================================
RCS file: /home/cvs/gateway/gwlib/conn.h,v
retrieving revision 1.26
diff -u -p -b -r1.26 conn.h
--- gwlib/conn.h	15 Nov 2003 13:14:23 -0000	1.26
+++ gwlib/conn.h	24 Nov 2003 10:34:18 -0000
@@ -220,7 +220,8 @@ int conn_wait(Connection *conn, double s
  * is done, or until the thread is interrupted or woken up.  Return 0
  * if it worked, 1 if there was an interruption, or -1 if the connection
  * is broken. */
-int conn_flush(Connection *conn);
+int conn_flush_real(Connection *conn, double tout_seconds);
+#define conn_flush(conn) conn_flush_real(conn, -1.0)
 
 /* Output functions.  Each of these takes an open connection and some
  * data, formats the data and queues it for sending.  It may also