various small things. type cleanups, loop over the burst point to make sure

it is at the right point. kick off the YP 'add' 5 seconds after source startup
so that any stats are processed.

svn path=/icecast/trunk/icecast/; revision=9314

src/cfgfile.c

@@ -32,6 +32,7 @@
 #define CONFIG_DEFAULT_CLIENT_LIMIT 256
 #define CONFIG_DEFAULT_SOURCE_LIMIT 16
 #define CONFIG_DEFAULT_QUEUE_SIZE_LIMIT (100*1024)
+#define CONFIG_DEFAULT_BURST_SIZE (64*1024)
 #define CONFIG_DEFAULT_THREADPOOL_SIZE 4
 #define CONFIG_DEFAULT_CLIENT_TIMEOUT 30
 #define CONFIG_DEFAULT_HEADER_TIMEOUT 15
@@ -361,7 +362,7 @@ static void _set_defaults(ice_config_t *configuration)
     configuration->relay_username = xmlStrdup (CONFIG_DEFAULT_MASTER_USERNAME);
     configuration->relay_password = NULL;
     /* default to a typical prebuffer size used by clients */
-    configuration->burst_size = 65536;
+    configuration->burst_size = CONFIG_DEFAULT_BURST_SIZE;
 }
 
 static void _parse_root(xmlDocPtr doc, xmlNodePtr node, 

src/connection.c

@@ -926,7 +926,7 @@ static void _handle_get_request (client_t *client, char *passed_uri)
          * the client, also.
          */
         if (source->max_listeners != -1 && 
-                source->listeners >= source->max_listeners) 
+                source->listeners >= (unsigned long)source->max_listeners) 
         {
             global_unlock();
             avl_tree_unlock(global.source_tree);

src/main.c

@@ -236,7 +236,7 @@ static int _start_logging(void)
     } else {
         playlistlog = -1;
     }
-    
+
     log_set_level(errorlog, config->loglevel);
     log_set_level(accesslog, 4);
     log_set_level(playlistlog, 4);

src/refbuf.h

@@ -21,7 +21,7 @@
 typedef struct _refbuf_tag
 {
     char *data;
-    long len;
+    unsigned long len;
     int sync_point;
     struct _refbuf_tag *associated;
     struct _refbuf_tag *next;

src/slave.c

@@ -508,6 +508,8 @@ static void *_slave_thread(void *arg)
     ice_config_t *config;
     unsigned int interval = 0;
 
+    source_recheck_mounts();
+
     while (1)
     {
         relay_server *cleanup_relays;

src/source.c

@@ -434,7 +434,7 @@ static refbuf_t *get_next_buffer (source_t *source)
         {
             if (source->last_read + (time_t)source->timeout < current)
             {
-                DEBUG3 ("last %ld, timeout %ld, now %ld", source->last_read, source->timeout, current);
+                DEBUG3 ("last %ld, timeout %d, now %ld", source->last_read, source->timeout, current);
                 WARN0 ("Disconnecting source due to socket timeout");
                 source->running = 0;
             }
@@ -655,14 +655,18 @@ void source_main (source_t *source)
 
             /* new data on queue, so check the burst point */
             source->burst_offset += refbuf->len;
-            if (source->burst_offset > source->burst_size)
+            while (source->burst_offset > source->burst_size)
             {
-                if (source->burst_point->next)
+                refbuf_t *to_release = source->burst_point;
+
+                if (to_release->next)
                 {
-                    refbuf_release (source->burst_point);
-                    source->burst_point = source->burst_point->next;
-                    source->burst_offset -= source->burst_point->len;
+                    source->burst_point = to_release->next;
+                    source->burst_offset -= to_release->len;
+                    refbuf_release (to_release);
+                    continue;
                 }
+                break;
             }
 
             /* save stream to file */
@@ -701,7 +705,7 @@ void source_main (source_t *source)
         while (client_node) {
 
             if(source->max_listeners != -1 && 
-                    source->listeners >= source->max_listeners) 
+                    source->listeners >= (unsigned long)source->max_listeners) 
             {
                 /* The common case is caught in the main connection handler,
                  * this deals with rarer cases (mostly concerning fallbacks)
@@ -740,7 +744,7 @@ void source_main (source_t *source)
         /* update the stats if need be */
         if (source->listeners != listeners)
         {
-            INFO2("listener count on %s now %d", source->mount, source->listeners);
+            INFO2("listener count on %s now %ld", source->mount, source->listeners);
             stats_event_args (source->mount, "listeners", "%d", source->listeners);
         }
 
@@ -875,6 +879,7 @@ static void _parse_audio_info (source_t *source, const char *s)
 }
 
 
+/* Apply the mountinfo details to the source */
 static void source_apply_mount (source_t *source, mount_proxy *mountinfo)
 {
     DEBUG1("Applying mount information for \"%s\"", source->mount);
@@ -951,7 +956,8 @@ void source_update_settings (ice_config_t *config, source_t *source)
         snprintf (buf, sizeof (buf), "%lu", source->max_listeners);
         stats_event (source->mount, "max_listeners", buf);
     }
-    DEBUG1 ("max listeners to %d", source->max_listeners);
+    DEBUG1 ("public set to %d", source->yp_public);
+    DEBUG1 ("max listeners to %ld", source->max_listeners);
     DEBUG1 ("queue size to %u", source->queue_size_limit);
     DEBUG1 ("burst size to %u", source->burst_size);
     DEBUG1 ("source timeout to %u", source->timeout);
@@ -966,22 +972,23 @@ void *source_client_thread (void *arg)
 
     source->client->respcode = 200;
     bytes = sock_write_bytes (source->client->con->sock, ok_msg, sizeof (ok_msg)-1);
-    if (bytes < sizeof (ok_msg)-1)
+    if (bytes < (int)(sizeof (ok_msg)-1))
     {
         global_lock();
         global.sources--;
         global_unlock();
         WARN0 ("Error writing 200 OK message to source client");
+        source_free_source (source);
+        return NULL;
     }
-    else
-    {
-        source->client->con->sent_bytes += bytes;
 
-        stats_event_inc(NULL, "source_client_connections");
-        source_main (source);
-    }
+    stats_event_inc(NULL, "source_client_connections");
+    stats_event (source->mount, "listeners", "0");
+
+    source_main (source);
+
     source_free_source (source);
-    slave_rebuild_mounts ();
+    source_recheck_mounts ();
 
     return NULL;
 }

src/source.h

@@ -49,7 +49,7 @@ typedef struct source_tag
     char *dumpfilename; /* Name of a file to dump incoming stream to */
     FILE *dumpfile;
 
-    long listeners;
+    unsigned long listeners;
     long max_listeners;
     int yp_public;
     int yp_prevent;

src/yp.c

@@ -515,18 +515,11 @@ static ypdata_t *create_yp_entry (source_t *source)
             snprintf (url, ret, "http://%s:%d%s", config->hostname, config->port, source->mount);
         }
 
-        mountproxy = config->mounts;
-        while (mountproxy) {
-            if (strcmp (mountproxy->mountname, source->mount) == 0) {
-                if (mountproxy->cluster_password) {
-                   add_yp_info (yp, "cluster_password", 
-                       mountproxy->cluster_password, YP_CLUSTER_PASSWORD);
-                }
-                break;
-            }
-            mountproxy = mountproxy->next;
-        }
+        mountproxy = config_find_mount (config, source->mount);
+        if (mountproxy && mountproxy->cluster_password)
+            add_yp_info (yp, "cluster_password", mountproxy->cluster_password, YP_CLUSTER_PASSWORD);
         config_release_config();
+
         yp->listen_url = util_url_escape (url);
         free (url);
         if (yp->listen_url == NULL)
@@ -896,6 +889,7 @@ void yp_add (source_t *source)
             yp->server = server;
             yp->touch_interval = server->touch_interval;
             yp->next = server->pending_mounts;
+            yp->next_update = time(NULL) + 5;
             server->pending_mounts = yp;
             yp_update = 1;
         }
@@ -903,7 +897,6 @@ void yp_add (source_t *source)
     }
     thread_mutex_unlock (&yp_pending_lock);
     thread_rwlock_unlock (&yp_lock);
-    /* DEBUG1 ("Added %s to YP ", source->mount); */
 }