[PATCH 18/31] connection_process takes node, con_q_t gets refbuf, and con_t timeout, util updated

This allows all to be re-entrant.

Signed-off-by: Niv Sardi <[email protected]>
---
 src/connection.c |   27 ++++++++++++++++++---------
 src/connection.h |    1 +
 src/util.c       |   48 ++++++++++++++++++++++++++++++++++--------------
 3 files changed, 53 insertions(+), 23 deletions(-)

diff --git a/src/connection.c b/src/connection.c
index 118258d..9532bab 100644
--- a/src/connection.c
+++ b/src/connection.c
@@ -84,6 +84,7 @@
 
 typedef struct connection_queue_tag {
     connection_t *con;
+    refbuf_t *refbuf;
     struct connection_queue_tag *next;
 } connection_queue_t;
 
@@ -115,7 +116,7 @@ rwlock_t _source_shutdown_rwlock;
 
 static void _handle_shoutcast_compatible (int shoutcast, char *shoutcast_mount);
 static int _handle_client (client_t *client);
-static int _connection_process (connection_t *con, int timeout);
+static int _connection_process (connection_queue_t *node);
 static void *_connection_thread (void *arg);
 
 static int compare_ip (void *arg, void *a, void *b)
@@ -633,7 +634,7 @@ static void *_connection_thread (void *arg)
         if (!node) {
             continue;
         }
-	err = _connection_process (node, 3000);
+	err = _connection_process (node);
         if (err > 0) {
             free(node);
             continue;
@@ -662,18 +663,20 @@ static void *_connection_thread (void *arg)
     return NULL;
 }
 
-static int _connection_process (connection_t *con, int timeout) {
+static int _connection_process (connection_queue_t *node) {
     ice_config_t *config;
     client_t *client = NULL;
     listener_t *listener;
-    refbuf_t *header = NULL;
+    refbuf_t *header;
     http_parser_t *parser = NULL;
     int hdrsize = 0;
     int shoutcast = 0;
     char *shoutcast_mount = NULL;
 
-    header = refbuf_new (PER_CLIENT_REFBUF_SIZE);
-    hdrsize = util_read_header (con, header, HEADER_READ_ENTIRE);
+    if (!node->refbuf)
+	    node->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
+    header = node->refbuf;
+    hdrsize = util_read_header (node->con, header, HEADER_READ_ENTIRE);
     if (hdrsize < 0)
     {
         global_unlock();
@@ -700,7 +703,7 @@ static int _connection_process (connection_t *con, int timeout) {
 
     if (header->sync_point && (parser->req_type == httpp_req_source ||
                                parser->req_type == httpp_req_post)) {
-	    hdrsize = util_read_header (con, header, HEADER_READ_ENTIRE);
+	    hdrsize = util_read_header (node->con, header, HEADER_READ_ENTIRE);
 	    if (hdrsize < 0) {
             INFO ("Header read failed");
             return hdrsize;
@@ -708,7 +711,7 @@ static int _connection_process (connection_t *con, int timeout) {
     }
 
     global_lock();
-    if (client_create (&client, con, parser) < 0)
+    if (client_create (&client, node->con, parser) < 0)
     {
         global_unlock();
         client_send_403 (client, "Icecast connection limit reached");
@@ -745,8 +748,12 @@ static int _connection_process (connection_t *con, int timeout) {
     global_unlock();
     config_release_config();
 
-    if (client->con->con_time + timeout <= time(NULL))
+/* XXX(xaiki): this should be 1, but actually, it's buggy, the client is already up and all.. */
+    if (client->con->con_timeout <= time(NULL)) {
+        WARN("there might be a bug if you see this");
+        client_destroy (client);
         return -1;
+    }
 
     stats_event_inc (NULL, "connections");
 
@@ -779,6 +786,8 @@ void connection_accept_loop (void)
             continue;
         }
 
+        con->con_timeout = time(NULL) + timeout;
+
         /* add connection async to the connection queue, then the
          * connection loop will do all the dirty work */
         node =_connection_node_new (con);
diff --git a/src/connection.h b/src/connection.h
index 80a2b10..42483cc 100644
--- a/src/connection.h
+++ b/src/connection.h
@@ -34,6 +34,7 @@ typedef struct connection_tag
     unsigned long id;
 
     time_t con_time;
+    time_t con_timeout;
     time_t discon_time;
     uint64_t sent_bytes;
 
diff --git a/src/util.c b/src/util.c
index 94b06f6..c74d1ce 100644
--- a/src/util.c
+++ b/src/util.c
@@ -142,9 +142,18 @@ int util_read_header(connection_t *con, refbuf_t *refbuf, int flags)
         return -ENOENT;
     }
 
-    config = config_get_config();
-    header_timeout = config->header_timeout*1000;
-    config_release_config();
+    if (con->con_timeout == 0) {
+        DEBUG0 ("NO TIMEOUT");
+        config = config_get_config();
+        header_timeout = config->header_timeout*1000;
+        config_release_config();
+    } else if (time(NULL) < con->con_timeout) {
+        DEBUG3 ("STILL HAVE TIME (%d - %d = %d)", time(NULL), con->con_timeout, con->con_timeout - time(NULL));
+        header_timeout = 1000;
+    } else {
+        DEBUG0 ("HUM BROKEN PIPE");
+        return -EPIPE;
+    }
 
     if (util_timed_wait_for_fd(con->sock, header_timeout) <= 0) {
 	    INFO("util_timed_wait_for_fd <= 0");
@@ -152,24 +161,19 @@ int util_read_header(connection_t *con, refbuf_t *refbuf, int flags)
     }
 
     if (refbuf->sync_point < 0) {
-        DEBUG ("REENTRING, got and old non-resolved sync");
+        DEBUG0 ("REENTRING, got and old non-resolved sync");
         pos = -refbuf->sync_point;
     } else if (refbuf->sync_point > 0) {
-        DEBUG ("REENTRING, got and old resolved sync");
+        DEBUG0 ("REENTRING, got and old resolved sync");
         endpos = pos = refbuf->sync_point;
     } else {
-        DEBUG ("FIRST TIME, no sync");
+        DEBUG0 ("FIRST TIME, no sync");
         pos = 0;
     }
 
-    while ((bytes = sock_read_bytes (con->sock, refbuf->data + pos, refbuf->len - pos)) >= 0) {
-        if (bytes == 0)
-            con->error = 1;
-        if (bytes == -1 && !sock_recoverable (sock_error()))
-            con->error = 1;
-
-        DEBUG("read %d, %d '%s'\nfrom pos '%s'", bytes, endpos, refbuf->data, refbuf->data + pos);
-	/* this is used for re-entrance, so we get a new chance to read */
+    while ((bytes = sock_read_bytes (con->sock, refbuf->data + pos, refbuf->len - pos)) > 0) {
+        DEBUG4 ("read %d, %d '%s'\nfrom pos '%s'", bytes, endpos, refbuf->data, refbuf->data + pos);
+        /* this is used for re-entrance, so we get a new chance to read */
         if (endpos == -ENOENT)
             endpos = util_find_eos_delim (refbuf, -(bytes + pos), flags);
         if (endpos != -ENOENT) {
@@ -198,8 +202,24 @@ int util_read_header(connection_t *con, refbuf_t *refbuf, int flags)
             refbuf->sync_point = -pos;
 		    return -EAGAIN;
 	    }
+
+        if (time(NULL) > con->con_timeout)
+            goto out_FAIL;
+    }
+
+    if (bytes == 0 || ! sock_recoverable (sock_error())) {
+        WARN ("Connection error");
+        con->error = 1;
+        return -EPIPE;
+    }
+
+    if (time(NULL) < con->con_timeout) {
+        DEBUG3 ("STILL HAVE TIME pos == (%d)…bytes = %d… data is %s", pos, bytes, refbuf->data);
+        DEBUG3 ("OUT, %p, refbuf->len = %d, refbuf->syncpoint = %d", refbuf, refbuf->len, refbuf->sync_point);
+        return -EAGAIN;
     }
 
+out_FAIL:
     WARN("Couldn't find enough data, pos = %d, data = %s, entire = %d.\n", pos, refbuf->data, flags);
     refbuf->sync_point = 0;
     return -ENOENT;
-- 
1.7.1

_______________________________________________
Icecast-dev mailing list
[email protected]
http://lists.xiph.org/mailman/listinfo/icecast-dev