new feature in icecast : burst-on-connect - allows an initial burst of data to...

new feature in icecast : burst-on-connect - allows an initial burst of data to connecting listeners, thus reducing the startup time of a stream.

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

conf/icecast.xml.in

@@ -7,6 +7,9 @@
         <client-timeout>30</client-timeout>
         <header-timeout>15</header-timeout>
         <source-timeout>10</source-timeout>
+        <!-- This will provide a burst of data when a client first connects,
+             thereby significantly reducing the startup time for a listener -->
+        <burst-on-connect>1</burst-on-connect>
     </limits>
 
     <authentication>

doc/icecast2_config_file.html

@@ -31,6 +31,7 @@ This section will describe each section of the config file and is grouped into t
         <client-timeout>30<client-timeout>
         <header-timeout>15<header-timeout>
         <source-timeout>10<source-timeout>
+        <burst-on-connect>1<burst-on-connect>
     <limits>
 </pre>
 <p>This section contains server level settings that, in general, do not need to be changed.  Only modify this section if you are know what you are doing.
@@ -63,6 +64,10 @@ The maximum time (in seconds) to wait for a request to come in once the client h
 <div class=indentedbox>
 If a connected source does not send any data within this timeout period (in seconds), then the source connection will be removed from the server.
 </div>
+<h4>burst-on-connect</h4>
+<div class=indentedbox>
+With this enabled, a connecting client will be sent a burst of audio data from the stream.  This will have the effect of reducing the startup time for the stream from the perspective of the listener.  This is due to the fact that most media players have local buffers that must be filled before the stream begins to play.  This may introduce a small latency in the stream (difference in time between when the source plays a clip and the listener hears a clip).  If this latency is important to you, then you can disable this feature.  The latency is bitrate-dependent, but as an example, for a 128kbps stream, the latency between the source and the player is ~ 1.5 secs WITHOUT burst on connect, and WITH burst on connect the latency is 3 secs.
+</div>
 <br>
 <br>
 <br>

src/cfgfile.c

@@ -339,6 +339,7 @@ static void _set_defaults(ice_config_t *configuration)
     configuration->num_yp_directories = 0;
     configuration->relay_username = NULL;
     configuration->relay_password = NULL;
+    configuration->burst_on_connect = 1;
 }
 
 static void _parse_root(xmlDocPtr doc, xmlNodePtr node, 
@@ -460,6 +461,10 @@ static void _parse_limits(xmlDocPtr doc, xmlNodePtr node,
             tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
             configuration->source_timeout = atoi(tmp);
             if (tmp) xmlFree(tmp);
+        } else if (strcmp(node->name, "burst-on-connect") == 0) {
+            tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
+            configuration->burst_on_connect = atoi(tmp);
+            if (tmp) xmlFree(tmp);
         }
     } while ((node = node->next));
 }

src/cfgfile.h

@@ -133,6 +133,7 @@ typedef struct ice_config_tag
     char *yp_url[MAX_YP_DIRECTORIES];
     int    yp_url_timeout[MAX_YP_DIRECTORIES];
     int num_yp_directories;
+    int burst_on_connect;
 } ice_config_t;
 
 typedef struct {

src/client.c

@@ -41,6 +41,7 @@ client_t *client_create(connection_t *con, http_parser_t *parser)
     client->parser = parser;
     client->queue = NULL;
     client->pos = 0;
+    client->burst_sent = 0;
 
     return client;
 }

src/client.h

@@ -45,6 +45,7 @@ typedef struct _client_tag
 
     /* function to call to release format specific resources */
     void (*free_client_data)(struct _client_tag *client);
+    int burst_sent;
 } client_t;
 
 client_t *client_create(connection_t *con, http_parser_t *parser);

src/format_vorbis.c

@@ -119,10 +119,10 @@ int format_vorbis_get_buffer(format_plugin_t *self, char *data, unsigned long le
     int i, result;
     ogg_packet op;
     char *tag;
-    refbuf_t *refbuf;
+    refbuf_t *refbuf, *source_refbuf;
     vstate_t *state = (vstate_t *)self->_state;
-#ifdef USE_YP
     source_t *source;
+#ifdef USE_YP
     time_t current_time;
 #endif
 
@@ -185,13 +185,24 @@ int format_vorbis_get_buffer(format_plugin_t *self, char *data, unsigned long le
                 ogg_stream_clear(&state->os);
                 vorbis_comment_clear(&state->vc);
                 vorbis_info_clear(&state->vi);
-#ifdef USE_YP
-                /* If we get an update on the mountpoint, force a
-                   yp touch */
+
+                /* Drain the source queue on metadata update otherwise you
+                   could have a mismatch between what is on the source queue
+                   and what is in the state->headbuf */
                 avl_tree_rlock(global.source_tree);
                 source = source_find_mount_raw(self->mount);
                 avl_tree_unlock(global.source_tree);
 
+                thread_mutex_lock(&source->queue_mutex);
+                while ((source_refbuf = refbuf_queue_remove(&source->queue))) {
+                    refbuf_release(source_refbuf);
+                }
+                thread_mutex_unlock(&source->queue_mutex);
+
+#ifdef USE_YP
+                /* If we get an update on the mountpoint, force a
+                   yp touch */
+
                 if (source) {
                     /* If we get an update on the mountpoint, force a
                        yp touch */

src/refbuf.c

@@ -101,6 +101,20 @@ refbuf_t *refbuf_queue_remove(refbuf_queue_t **queue)
        
     return refbuf;
 }
+refbuf_t * refbuf_queue_get(refbuf_queue_t **queue, int item)
+{   
+    refbuf_queue_t *node = *queue;
+    int size = 0; 
+    while (node) {
+        if (size == item) {
+            return node->refbuf;
+        }
+        node = node->next; 
+        size++;
+    }
+    return NULL;
+}
+
 
 void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf)
 {

src/refbuf.h

@@ -49,6 +49,7 @@ void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf);
 int refbuf_queue_size(refbuf_queue_t **queue);
 /* Size in bytes */
 int refbuf_queue_length(refbuf_queue_t **queue);
+refbuf_t * refbuf_queue_get(refbuf_queue_t **queue, int item);
 
 #endif  /* __REFBUF_H__ */
 

src/source.c

@@ -194,6 +194,7 @@ int source_compare_sources(void *arg, void *a, void *b)
 
 void source_clear_source (source_t *source)
 {
+    refbuf_t *refbuf;
 #ifdef USE_YP
     int i;
 #endif
@@ -254,6 +255,12 @@ void source_clear_source (source_t *source)
 
     free(source->dumpfilename);
     source->dumpfilename = NULL;
+    /* Lets clear out the source queue too */
+    while ((refbuf = refbuf_queue_remove(&source->queue)))
+        refbuf_release(refbuf);
+    source->queue = NULL;
+    source->burst_on_connect = 1;
+    thread_mutex_destroy(&source->queue_mutex);
 }
 
 
@@ -484,6 +491,7 @@ static void source_init (source_t *source)
     memset (listenurl, '\000', listen_url_size);
     snprintf (listenurl, listen_url_size, "http://%s:%d%s",
             config->hostname, config->port, source->mount);
+    source->burst_on_connect = config->burst_on_connect;
     config_release_config();
 
     stats_event (source->mount, "listenurl", listenurl);
@@ -514,6 +522,8 @@ static void source_init (source_t *source)
 
     sock_set_blocking (source->con->sock, SOCK_NONBLOCK);
 
+    thread_mutex_create(&source->queue_mutex);
+
     DEBUG0("Source creation complete");
     source->running = 1;
 
@@ -543,11 +553,11 @@ void source_main (source_t *source)
 {
     char buffer[4096];
     long bytes, sbytes;
-    int ret;
+    int ret, i;
     client_t *client;
     avl_node *client_node;
 
-    refbuf_t *refbuf, *abuf;
+    refbuf_t *refbuf, *abuf, *stale_refbuf;
     int data_done;
 
     source_init (source);
@@ -558,6 +568,27 @@ void source_main (source_t *source)
             WARN0("Bad data from source");
             break;
         }
+        if (source->burst_on_connect) {
+            thread_mutex_lock(&source->queue_mutex);
+            /* Add to the source buffer */
+            if (refbuf) {
+                refbuf_addref(refbuf);
+                refbuf_queue_add(&(source->queue), refbuf);
+                /* We derive the size of the source buffer queue based off the 
+                   setting for queue_size_limit (client buffer queue size).  
+                   This is because the source buffer queue size should be a
+                   percentage of the client buffer size (definately cannot
+                   be larger). Why 50% ? Because > 75% does not give the
+                   client enough leeway for lagging on initial connection
+                   and < 25% does not provide a good enough burst on connect. */
+                if (refbuf_queue_length(&(source->queue)) > 
+                    source->queue_size_limit/2) {
+                    stale_refbuf = refbuf_queue_remove(&(source->queue));
+                    refbuf_release(stale_refbuf);
+                }
+            }
+            thread_mutex_unlock(&source->queue_mutex);
+        }
         bytes = 1; /* Set to > 0 so that the post-loop check won't be tripped */
         while (refbuf == NULL) {
             bytes = 0;
@@ -590,6 +621,20 @@ void source_main (source_t *source)
                 WARN0("Bad data from source");
                 goto done;
             }
+            if (source->burst_on_connect) {
+                /* Add to the source buffer */
+                thread_mutex_lock(&source->queue_mutex);
+                if (refbuf) {
+                    refbuf_addref(refbuf);
+                    refbuf_queue_add(&(source->queue), refbuf);
+                    if (refbuf_queue_length(&(source->queue)) > 
+                        source->queue_size_limit/2) {
+                        stale_refbuf = refbuf_queue_remove(&(source->queue));
+                        refbuf_release(stale_refbuf);
+                    }
+                }
+                thread_mutex_unlock(&source->queue_mutex);
+            }
         }
 
         if (bytes <= 0) {
@@ -710,7 +755,10 @@ void source_main (source_t *source)
         /* release read lock on client_tree */
         avl_tree_unlock(source->client_tree);
 
-        refbuf_release(refbuf);
+        /* Only release the refbuf if we didn't add it to the source queue */
+        if (!source->burst_on_connect) {
+            refbuf_release(refbuf);
+        }
 
         /* acquire write lock on client_tree */
         avl_tree_wlock(source->client_tree);
@@ -767,10 +815,26 @@ void source_main (source_t *source)
             /* we have to send cached headers for some data formats
             ** this is where we queue up the buffers to send
             */
+            client = (client_t *)client_node->key;
             if (source->format->has_predata) {
-                client = (client_t *)client_node->key;
                 client->queue = source->format->get_predata(source->format);
             }
+            if (source->burst_on_connect) {
+                /* here is where we fill up the new client with refbufs from
+                   the source buffer.  this will allow an initial burst of
+                   audio data to be sent to the client, and allow for a faster
+                   startup time (from listener perspective) for the stream */ 
+                if (!client->burst_sent) {
+                    thread_mutex_lock(&source->queue_mutex);
+                    for (i=0;i<refbuf_queue_size(&(source->queue));i++) {
+                        refbuf_queue_add(&(client->queue),
+                            refbuf_queue_get(&(source->queue), i));
+                    }
+                    thread_mutex_unlock(&source->queue_mutex);
+                    client->burst_sent = 1;
+                    DEBUG1("Added %d buffers to initial client queue", refbuf_queue_length(&(source->queue)));
+                }
+            }
 
             client_node = avl_get_next(client_node);
         }

src/source.h

@@ -58,6 +58,9 @@ typedef struct source_tag
     int no_mount;
     unsigned queue_size_limit;
     unsigned timeout;  /* source timeout in seconds */
+    refbuf_queue_t *queue;
+    mutex_t queue_mutex;
+    int burst_on_connect;
 } source_t;
 
 source_t *source_reserve (const char *mount);