[PATCH] http client timeouts part1 (v3)

Index: gwlib/conn.c
===================================================================
RCS file: /home/cvs/gateway/gwlib/conn.c,v
retrieving revision 1.72
diff -a -u -p -r1.72 conn.c
--- gwlib/conn.c	18 Feb 2004 16:52:33 -0000	1.72
+++ gwlib/conn.c	18 Nov 2004 20:38:34 -0000
@@ -139,6 +139,7 @@ struct Connection
     FDSet *registered;
     conn_callback_t *callback;
     void *callback_data;
+    conn_callback_data_destroyer_t *callback_data_destroyer;
     /* Protected by inlock */
     int listening_pollin;
     /* Protected by outlock */
@@ -166,7 +167,7 @@ static void unlocked_register_pollout(Co
 #define unlock_out(conn) unlock_out_real(conn, __FILE__, __LINE__, __func__)
 
 /* Lock a Connection's read direction, if the Connection is unclaimed */
-static void lock_in(Connection *conn)
+static void inline lock_in(Connection *conn)
 {
     gw_assert(conn != NULL);
 
@@ -177,23 +178,21 @@ static void lock_in(Connection *conn)
 }
 
 /* Unlock a Connection's read direction, if the Connection is unclaimed */
-static void unlock_in_real(Connection *conn, char *file, int line, const char *func)
+static void inline unlock_in_real(Connection *conn, char *file, int line, const char *func)
 {
     int ret;
     gw_assert(conn != NULL);
 
-    if (!conn->claimed) {
-        if ((ret = mutex_unlock(conn->inlock)) != 0) {
-            panic(0, "%s:%ld: %s: Mutex unlock failed. " \
-		             "(Called from %s:%ld:%s.)", \
-			         __FILE__, (long) __LINE__, __func__, \
-			         file, (long) line, func);
-        }
-     }
+    if (!conn->claimed && (ret = mutex_unlock(conn->inlock)) != 0) {
+        panic(0, "%s:%ld: %s: Mutex unlock failed. "
+            "(Called from %s:%ld:%s.)",
+            __FILE__, (long) __LINE__, __func__,
+            file, (long) line, func);
+    }
 }
 
 /* Lock a Connection's write direction, if the Connection is unclaimed */
-static void lock_out(Connection *conn)
+static void inline lock_out(Connection *conn)
 {
     gw_assert(conn != NULL);
 
@@ -204,29 +203,27 @@ static void lock_out(Connection *conn)
 }
 
 /* Unlock a Connection's write direction, if the Connection is unclaimed */
-static void unlock_out_real(Connection *conn, char *file, int line, const char *func)
+static void inline unlock_out_real(Connection *conn, char *file, int line, const char *func)
 {
     int ret;
     gw_assert(conn != NULL);
 
-    if (!conn->claimed) {
-        if ((ret = mutex_unlock(conn->outlock)) != 0) {
-            panic(0, "%s:%ld: %s: Mutex unlock failed. " \
-		             "(Called from %s:%ld:%s.)", \
-			         __FILE__, (long) __LINE__, __func__, \
-			         file, (long) line, func);
-        }
-     }
+    if (!conn->claimed && (ret = mutex_unlock(conn->outlock)) != 0) {
+        panic(0, "%s:%ld: %s: Mutex unlock failed. "
+            "(Called from %s:%ld:%s.)",
+            __FILE__, (long) __LINE__, __func__,
+            file, (long) line, func);
+    }
 }
 
 /* Return the number of bytes in the Connection's output buffer */
-static long unlocked_outbuf_len(Connection *conn)
+static long inline unlocked_outbuf_len(Connection *conn)
 {
     return octstr_len(conn->outbuf) - conn->outbufpos;
 }
 
 /* Return the number of bytes in the Connection's input buffer */
-static long unlocked_inbuf_len(Connection *conn)
+static long inline unlocked_inbuf_len(Connection *conn)
 {
     return octstr_len(conn->inbuf) - conn->inbufpos;
 }
@@ -551,6 +548,7 @@ Connection *conn_wrap_fd(int fd, int ssl
     conn->registered = NULL;
     conn->callback = NULL;
     conn->callback_data = NULL;
+    conn->callback_data_destroyer = NULL;
     conn->listening_pollin = 0;
     conn->listening_pollout = 0;
 #ifdef HAVE_LIBSSL
@@ -595,8 +593,12 @@ void conn_destroy(Connection *conn)
     /* No locking done here.  conn_destroy should not be called
      * if any thread might still be interested in the connection. */
 
-    if (conn->registered)
+    if (conn->registered) {
         fdset_unregister(conn->registered, conn->fd);
+        /* call data destroyer if any */
+        if (conn->callback_data != NULL && conn->callback_data_destroyer != NULL)
+            conn->callback_data_destroyer(conn->callback_data);
+    }
 
     if (conn->fd >= 0) {
         /* Try to flush any remaining data */
@@ -674,8 +676,8 @@ int conn_error(Connection *conn)
 {
     int err;
 
-    lock_in(conn);
     lock_out(conn);
+    lock_in(conn);
     err = conn->io_error;
     unlock_in(conn);
     unlock_out(conn);
@@ -721,8 +723,8 @@ static void poll_callback(int fd, int re
      * fdset and set the error condition variable to let the upper layer
      * close and destroy the connection. */
     if (revents & (POLLERR|POLLHUP)) {
-        lock_in(conn);
         lock_out(conn);
+        lock_in(conn);
         if (conn->listening_pollin)
             unlocked_register_pollin(conn, 0);
         if (conn->listening_pollout)
@@ -756,8 +758,8 @@ static void poll_callback(int fd, int re
         conn->callback(conn, conn->callback_data);
 }
 
-int conn_register(Connection *conn, FDSet *fdset,
-                  conn_callback_t callback, void *data)
+int conn_register_real(Connection *conn, FDSet *fdset,
+                  conn_callback_t callback, void *data, conn_callback_data_destroyer_t *data_destroyer)
 {
     int events;
     int result = 0;
@@ -775,7 +777,11 @@ int conn_register(Connection *conn, FDSe
     if (conn->registered == fdset) {
         /* Re-registering.  Change only the callback info. */
         conn->callback = callback;
+        /* call data destroyer if new data supplied */
+        if (conn->callback_data != NULL && conn->callback_data != data && conn->callback_data_destroyer != NULL)
+            conn->callback_data_destroyer(conn->callback_data);
         conn->callback_data = data;
+        conn->callback_data_destroyer = data_destroyer;
         result = 0;
     } else if (conn->registered) {
         /* Already registered to a different fdset. */
@@ -795,14 +801,15 @@ int conn_register(Connection *conn, FDSe
         conn->registered = fdset;
         conn->callback = callback;
         conn->callback_data = data;
+        conn->callback_data_destroyer = data_destroyer;
         conn->listening_pollin = (events & POLLIN) != 0;
         conn->listening_pollout = (events & POLLOUT) != 0;
         fdset_register(fdset, conn->fd, events, poll_callback, conn);
         result = 0;
     }
 
-    unlock_out(conn);
     unlock_in(conn);
+    unlock_out(conn);
 
     return result;
 }
@@ -822,7 +829,11 @@ void conn_unregister(Connection *conn)
         fdset_unregister(conn->registered, conn->fd);
         conn->registered = NULL;
         conn->callback = NULL;
+        /* call data destroyer */
+        if (conn->callback_data != NULL && conn->callback_data_destroyer != NULL)
+            conn->callback_data_destroyer(conn->callback_data);
         conn->callback_data = NULL;
+        conn->callback_data_destroyer = NULL;
         conn->listening_pollin = 0;
         conn->listening_pollout = 0;
     }
@@ -1173,12 +1184,13 @@ Octstr *conn_read_packet(Connection *con
 X509 *conn_get_peer_certificate(Connection *conn) 
 {
     /* Don't know if it needed to be locked , but better safe as crash */
-    lock_in(conn);
     lock_out(conn);
+    lock_in(conn);
     if (conn->peer_certificate == NULL && conn->ssl != NULL)
         conn->peer_certificate = SSL_get_peer_certificate(conn->ssl);
     unlock_in(conn);
     unlock_out(conn);
+    
     return conn->peer_certificate;
 }
 
Index: gwlib/conn.h
===================================================================
RCS file: /home/cvs/gateway/gwlib/conn.h,v
retrieving revision 1.29
diff -a -u -p -r1.29 conn.h
--- gwlib/conn.h	16 Feb 2004 19:41:26 -0000	1.29
+++ gwlib/conn.h	18 Nov 2004 20:38:34 -0000
@@ -104,6 +104,14 @@ typedef struct Connection Connection;
  * down the polling process.  This may be good or bad. */
 typedef void conn_callback_t(Connection *conn, void *data);
 
+/*
+ * If conn_register was called for this connection, a callback data destroyer
+ * function will be called if conn_unregister, conn_destroy or conn_register
+ * (with different data) called for this connection.
+ * This function is responsible to destroy callback data.
+ */   
+typedef void conn_callback_data_destroyer_t(void *data);
+
 #ifdef HAVE_LIBSSL
 /* Open an SSL connection to the given host and port.  Same behavior
  * as conn_open_tcp() below. 'certkeyfile' specifies a PEM-encoded
@@ -198,10 +206,14 @@ void conn_set_output_buffering(Connectio
  * than calling conn_unregister first.
  * NOTE: Using conn_register will always mean that the Connection will be
  * used by more than one thread, so don't also call conn_claim. */
-int conn_register(Connection *conn, FDSet *fdset,
-                  conn_callback_t callback, void *data);
+#define conn_register(conn, fdset, callback, data) \
+    conn_register_real(conn, fdset, callback, data, NULL)
+int conn_register_real(Connection *conn, FDSet *fdset,
+    conn_callback_t callback, void *data, conn_callback_data_destroyer_t destroyer);
 
-/* Remove the current registration. */
+/*
+ * Remove the current registration and call data destroyer if not NULL.
+ */ 
 void conn_unregister(Connection *conn);
 
 /* Block the thread until one of the following is true:
Index: gwlib/http.c
===================================================================
RCS file: /home/cvs/gateway/gwlib/http.c,v
retrieving revision 1.219
diff -a -u -p -r1.219 http.c
--- gwlib/http.c	11 Aug 2004 16:41:29 -0000	1.219
+++ gwlib/http.c	18 Nov 2004 20:38:35 -0000
@@ -601,6 +601,25 @@ static int entity_read(HTTPEntity *ent, 
  * HTTP client interface.
  */
 
+/*
+ * Internal lists of completely unhandled requests and requests for which
+ * a request has been sent but response has not yet been read.
+ */
+static List *pending_requests = NULL;
+
+
+/*
+ * Have background threads been started?
+ */
+static Mutex *client_thread_lock = NULL;
+static volatile sig_atomic_t client_threads_are_running = 0;
+
+
+/*
+ * Set of all connections to all servers. Used with conn_register to
+ * do I/O on several connections with a single thread.
+ */
+static FDSet *client_fdset = NULL;
 
 /*
  * Maximum number of HTTP redirections to follow. Making this infinite
@@ -629,7 +648,7 @@ typedef struct {
     List *request_headers;
     Octstr *request_body;   /* NULL for GET or HEAD, non-NULL for POST */
     enum {
-      connecting,
+	connecting,
 	request_not_sent,
 	reading_status,
 	reading_entity,
@@ -707,17 +726,13 @@ static void server_destroy(void *p)
  * Pool of open, but unused connections to servers or proxies. Key is
  * "servername:port", value is List with Connection objects.
  */
-static Dict *conn_pool = NULL;
-static Mutex *conn_pool_lock = NULL;
+static Dict *conn_pool;
+static Mutex *conn_pool_lock;
 
 
 static void conn_pool_item_destroy(void *item)
 {
-    Connection *conn;
-    
-    while ((conn = list_extract_first(item)) != NULL)
-    	conn_destroy(conn);
-    list_destroy(item, NULL);
+    list_destroy(item, (void(*)(void*))conn_destroy);
 }
 
 static void conn_pool_init(void)
@@ -734,7 +749,7 @@ static void conn_pool_shutdown(void)
 }
 
 
-static Octstr *conn_pool_key(Octstr *host, int port)
+static inline Octstr *conn_pool_key(Octstr *host, int port)
 {
     return octstr_format("%S:%d", host, port);
 }
@@ -754,15 +769,18 @@ static Connection *conn_pool_get(Octstr 
     if (list == NULL)
     	conn = NULL;
     else {
-	while (1) {
-	    conn = list_extract_first(list);
-	    if (conn == NULL)
-		break;
-	    /* Check whether the server has closed the connection while
-	     * it has been in the pool. */
-	    conn_wait(conn, 0);
-	    if (!conn_eof(conn) && !conn_error(conn))
-		break;
+        while ((conn = list_extract_first(list)) != NULL) {
+#ifdef USE_KEEPALIVE
+            /* unregister our server disconnect callback */
+            conn_unregister(conn);
+#endif 
+            /*
+             * Check whether the server has closed the connection while
+             * it has been in the pool.
+             */
+            conn_wait(conn, 0);
+            if (!conn_eof(conn) && !conn_error(conn))
+                break;
 	    conn_destroy(conn);
 	}
     }
@@ -786,6 +804,40 @@ static Connection *conn_pool_get(Octstr 
 }
 
 #ifdef USE_KEEPALIVE
+static void check_pool_conn(Connection *conn, void *data)
+{
+    Octstr *key = data;
+    if (run_status != running) {
+        conn_unregister(conn);
+        return;
+    }
+    /* check if connection still ok */
+    conn_wait(conn, 0);
+    if (conn_error(conn) || conn_eof(conn)) {
+        List *list;
+        mutex_lock(conn_pool_lock);
+        list = dict_get(conn_pool, key);
+        if (list_delete_equal(list, conn) > 0) {
+            /*
+             * ok, connection was still within pool. So it's
+             * safe to destroy this connection.
+             */
+            debug("gwlib.http", 0, "HTTP: Server closed connection, destroying it <%s><%p>.",
+                  octstr_get_cstr(key), conn);
+            conn_destroy(conn);
+        } else {
+            /* hmm, bad... connections is already used */
+            error(0, "HTTP: Race condition detected: Could not found connection in pool!");
+            /*
+             * don't unregister or destroy our key.
+             * Better memleak as segfault ;)
+             */
+        }
+        mutex_unlock(conn_pool_lock);
+    }
+}
+
+
 static void conn_pool_put(Connection *conn, Octstr *host, int port)
 {
     Octstr *key;
@@ -799,33 +851,13 @@ static void conn_pool_put(Connection *co
         dict_put(conn_pool, key, list);
     }
     list_append(list, conn);
-    octstr_destroy(key);
+    /* register connection to get server disconnect */
+    conn_register_real(conn, client_fdset, check_pool_conn, key, octstr_destroy_item);
     mutex_unlock(conn_pool_lock);
 }
 #endif
 
 
-/*
- * Internal lists of completely unhandled requests and requests for which
- * a request has been sent but response has not yet been read.
- */
-static List *pending_requests = NULL;
-
-
-/*
- * Have background threads been started?
- */
-static Mutex *client_thread_lock = NULL;
-static volatile sig_atomic_t client_threads_are_running = 0;
-
-
-/*
- * Set of all connections to all servers. Used with conn_register to
- * do I/O on several connections with a single thread.
- */
-static FDSet *client_fdset = NULL;
-
-
 HTTPCaller *http_caller_create(void)
 {
     HTTPCaller *caller;
@@ -1025,7 +1057,7 @@ static void handle_transaction(Connectio
 
 #ifdef DUMP_RESPONSE
                 /* Dump the response */
-                debug("wsp.http", 0, "HTTP: Received response:");
+                debug("gwlib.http", 0, "HTTP: Received response:");
                 h = build_response(trans->response->headers, trans->response->body);
                 octstr_dump(h, 0);
                 octstr_destroy(h);
@@ -1051,11 +1083,11 @@ static void handle_transaction(Connectio
     if (trans->persistent) {
         if (proxy_used_for_host(trans->host))
             conn_pool_put(trans->conn, proxy_hostname, proxy_port);
-        else
+        else 
             conn_pool_put(trans->conn, trans->host, trans->port);
     } else
 #endif
-    	conn_destroy(trans->conn);
+        conn_destroy(trans->conn);
 
     trans->conn = NULL;
 
@@ -1422,13 +1454,11 @@ static void parse2trans(HTTPURLParse *p,
 
 static Connection *get_connection(HTTPServer *trans) 
 {
-    Connection *conn;
+    Connection *conn = NULL;
     Octstr *host;
     HTTPURLParse *p;
     int port;
 
-    conn = NULL;
-
     /* if the parsing has not yet been done, then do it now */
     if (!trans->host && trans->port == 0 && trans->url != NULL) {
         if ((p = parse_url(trans->url)) != NULL) {
@@ -1447,26 +1477,17 @@ static Connection *get_connection(HTTPSe
         port = trans->port;
     }
 
-    if (trans->retrying) {
-#ifdef HAVE_LIBSSL
-    if (trans->ssl) conn = conn_open_ssl(host, port, trans->certkeyfile, http_interface);
-        else
-#endif /* HAVE_LIBSSL */
-      conn = conn_open_tcp_nb(host, port, http_interface);
-            debug("gwlib.http", 0, "HTTP: Opening NEW connection to `%s:%d' (fd=%d).",
-                  octstr_get_cstr(host), port, conn_get_id(conn));
-    } else
     conn = conn_pool_get(host, port, trans->ssl, trans->certkeyfile,
                          http_interface);
     if (conn == NULL)
         goto error;
 
-  return conn;
+    return conn;
 
- error:
-  conn_destroy(conn);
-  error(0, "Couldn't send request to <%s>", octstr_get_cstr(trans->url));
-  return NULL;
+error:
+    conn_destroy(conn);
+    error(0, "Couldn't send request to <%s>", octstr_get_cstr(trans->url));
+    return NULL;
 }
 
 
@@ -1476,47 +1497,47 @@ static Connection *get_connection(HTTPSe
  */
 static int send_request(HTTPServer *trans)
 {
-  Octstr *request;
+    Octstr *request;
+
+    request = NULL;
 
-  request = NULL;
+    /* 
+    * we have to assume all values in trans are already set
+    * by parse_url() before calling this.
+    */
+
+    if (trans->username != NULL)
+        http_add_basic_auth(trans->request_headers, trans->username,
+                            trans->password);
 
-  /* 
-   * we have to assume all values in trans are already set
-   * by parse_url() before calling this.
-   */
-
-  if (trans->username != NULL)
-    http_add_basic_auth(trans->request_headers, trans->username,
-			trans->password);
-
-  if (proxy_used_for_host(trans->host)) {
-    proxy_add_authentication(trans->request_headers);
-    request = build_request(http_method2name(trans->method),
-			    trans->url, trans->host, trans->port, 
-			    trans->request_headers, 
-			    trans->request_body);
-  } else {
-    request = build_request(http_method2name(trans->method), trans->uri, 
-			    trans->host, trans->port,
-			    trans->request_headers,
-			    trans->request_body);
-  }
+    if (proxy_used_for_host(trans->host)) {
+        proxy_add_authentication(trans->request_headers);
+        request = build_request(http_method2name(trans->method),
+                                trans->url, trans->host, trans->port, 
+                                trans->request_headers, 
+                                trans->request_body);
+    } else {
+        request = build_request(http_method2name(trans->method), trans->uri, 
+                                trans->host, trans->port,
+                                trans->request_headers,
+                                trans->request_body);
+    }
   
-    debug("wsp.http", 0, "HTTP: Sending request:");
+    debug("gwlib.http", 0, "HTTP: Sending request:");
     octstr_dump(request, 0);
-  if (conn_write(trans->conn, request) == -1)
+    if (conn_write(trans->conn, request) == -1)
         goto error;
 
     octstr_destroy(request);
 
-  return 0;
+    return 0;
 
- error:
-  conn_destroy(trans->conn);
-  trans->conn = NULL;
+error:
+    conn_destroy(trans->conn);
+    trans->conn = NULL;
     octstr_destroy(request);
     error(0, "Couldn't send request to <%s>", octstr_get_cstr(trans->url));
-  return -1;
+    return -1;
 }
 
 
@@ -1544,46 +1565,43 @@ static void write_request_thread(void *a
          */
         trans->conn = get_connection(trans);
 
-	if (trans->conn == NULL)
-	  list_produce(trans->caller, trans);
-        else {
-          if (conn_is_connected(trans->conn) == 0) {
-	    debug("gwlib.http", 0, "Socket connected at once");
-
-        if (trans->method == HTTP_METHOD_POST) {
-            /* 
-             * Add a Content-Length header.  Override an existing one, if
-             * necessary.  We must have an accurate one in order to use the
-             * connection for more than a single request.
-             */
-            http_header_remove_all(trans->request_headers, "Content-Length");
-            sprintf(buf, "%ld", octstr_len(trans->request_body));
-            http_header_add(trans->request_headers, "Content-Length", buf);
-        } 
+        if (trans->conn == NULL)
+            list_produce(trans->caller, trans);
+        else if (conn_is_connected(trans->conn) == 0) {
+            debug("gwlib.http", 0, "Socket connected at once");
+
+            if (trans->method == HTTP_METHOD_POST) {
+                /* 
+                * Add a Content-Length header.  Override an existing one, if
+                * necessary.  We must have an accurate one in order to use the
+                * connection for more than a single request.
+                */
+                http_header_remove_all(trans->request_headers, "Content-Length");
+                sprintf(buf, "%ld", octstr_len(trans->request_body));
+                http_header_add(trans->request_headers, "Content-Length", buf);
+            } 
             /* 
              * ok, this has to be an GET or HEAD request method then,
              * if it contains a body, then this is not HTTP conform, so at
              * least warn the user 
              */
-        else if (trans->request_body != NULL) {
-            warning(0, "HTTP: GET or HEAD method request contains body:");
-            octstr_dump(trans->request_body, 0);
-        }
-	    if ((rc = send_request(trans)) == 0) {
-            trans->state = reading_status;
-              conn_register(trans->conn, client_fdset, handle_transaction, 
-                            trans);
+            else if (trans->request_body != NULL) {
+                warning(0, "HTTP: GET or HEAD method request contains body:");
+                octstr_dump(trans->request_body, 0);
+            }
+            if ((rc = send_request(trans)) == 0) {
+                trans->state = reading_status;
+                conn_register(trans->conn, client_fdset, handle_transaction, 
+                                trans);
             } else {
-              list_produce(trans->caller, trans);
+                list_produce(trans->caller, trans);
             }
 
-          } else { /* Socket not connected, wait for connection */
+        } else { /* Socket not connected, wait for connection */
             debug("gwlib.http", 0, "Socket connecting");
             trans->state = connecting;
             conn_register(trans->conn, client_fdset, handle_transaction, trans);
         }
-	  
-	}
     }
 }
 
@@ -1609,7 +1627,7 @@ static void start_client_threads(void)
 
 void http_set_interface(const Octstr *our_host)
 {
-  http_interface = octstr_duplicate(our_host);
+    http_interface = octstr_duplicate(our_host);
 }
 
 
@@ -2656,7 +2674,7 @@ Octstr *http_header_value(List *headers,
 List *http_header_duplicate(List *headers)
 {
     List *new;
-    long i;
+    long i, len;
 
     gwlib_assert_init();
 
@@ -2664,13 +2682,14 @@ List *http_header_duplicate(List *header
         return NULL;
 
     new = http_create_empty_headers();
-    for (i = 0; i < list_len(headers); ++i)
+    len = list_len(headers);
+    for (i = 0; i < len; ++i)
         list_append(new, octstr_duplicate(list_get(headers, i)));
     return new;
 }
 
 
-#define MAX_HEADER_LENGHT 256
+#define MAX_HEADER_LENGTH 256
 /*
  * Aggregate header in one (or more) lines with several parameters separated
  * by commas, instead of one header per parameter
@@ -2697,7 +2716,7 @@ void http_header_pack(List *headers)
             http_header_get(headers, j, &name2, &value2);
 
             if(octstr_case_compare(name, name2) == 0) {
-                if(octstr_len(value) + 2 + octstr_len(value2) > MAX_HEADER_LENGHT) {
+                if(octstr_len(value) + 2 + octstr_len(value2) > MAX_HEADER_LENGTH) {
 		    octstr_destroy(name2);
 		    octstr_destroy(value2);
                     break;
@@ -3278,8 +3297,8 @@ void http_shutdown(void)
 
     run_status = terminating;
 
-    conn_pool_shutdown();
     port_shutdown();
+    conn_pool_shutdown();
     client_shutdown();
     server_shutdown();
     proxy_shutdown();